vj_sim/sched/

kernel_adapter.rs

//! Task S4 (`docs/superpowers/plans/2026-09-26-wave-A2-slice.md`): the ONE
//! place in this tree a frozen-ABI compiled kernel (`vj_abi::ProcessFn`)
//! is bridged to `sched::CompiledProcess` -- ledger PR-100, PR-101, PR-102,
//! PR-103, PR-108, PR-109, PR-110, PR-112, PR-121, CG-083, CG-087, CG-088,
//! CG-089, CG-106, CG-107, CG-116, CG-164. See `docs/design/process.md` §4
//! and ARCHITECTURE.md §5/§6.4.
//!
//! ## `SV_KERNEL_ABI_RECONCILIATION` (CG-084/CG-111..CG-116 supersession)
//!
//! `docs/design/codegen.md` §9's literal `SvKernelTable`/`SvProcessDesc`/
//! `SvWakeSpec`/`SvResumeStatus` draft is superseded by ARCHITECTURE.md §5's
//! simpler, frozen ABI (`SimHandle`/`ProcessFn`/`EvalResult`/`Checkpoint`) --
//! the same reconciliation `docs/ledger.json`'s own `PR-037`/`PR-107` rows
//! already record for the process-model side. This module closes CG-084 and
//! CG-111 through CG-116 by implementing the frozen ABI shape below, not the
//! literal old struct names: [`CompiledKernelProcess`] IS what
//! `SvKernelTable`'s "resume a compiled process" ask reduces to once
//! `resume_pc` is a plain function argument instead of a state-arena-resident
//! discriminant slot (CG-088/CG-089's own reconciliation, applied at the
//! `vj-codegen::process_lower::lower_kernel_fn` call site that emits the
//! LLVM side of this same boundary).
//!
//! ## PR-121: `RescheduleReq`'s opaque bytes, given real meaning
//!
//! [`build_eval_result`]/[`decode_reschedule`] are the only two functions in
//! this tree that ever read or write `vj_abi::eval::RescheduleReq::opaque`
//! -- everywhere else treats it as an inert 32-byte blob. They reuse (never
//! reinvent) `super::reschedule_req_proposal`'s already-tested 24-byte tagged
//! union, packed into the frozen 32-byte `opaque` array alongside 4 more bytes
//! (`next_resume_pc`, little-endian) and 4 reserved zero bytes -- the "8 of
//! the 32 reserved bytes free for a future field" headroom that module's own
//! doc comment already sized in for.
//!
//! ## Known A2-slice scope limit -- CLOSED (A2-part2 review-fix r1)
//!
//! The frozen `ProcessFn = extern "C" fn(SimHandle, *mut u8, u32) ->
//! EvalResult` signature carries no `ProcessId`/`ProcessInstanceId` parameter
//! at all -- there is nowhere in this ABI boundary for a compiled kernel to
//! report "which process instance is this" to `nba_enqueue`'s own `source:
//! u32` parameter, so every `nba_enqueue` call this wave's codegen emits
//! still passes a constant `0` for `source` (`vj-codegen::process_lower`'s
//! own doc comment on its NBA lowering repeats this same note at the
//! call-emission site) -- an ABI-breaking 4th `ProcessFn` parameter remains
//! out of this task's authority (AB-050 is a cross-track-negotiated
//! contract). What this review fix closes is the OTHER half this section
//! used to flag as unbuilt: the runtime-side lookup keyed off "current
//! process" ambient state. [`CURRENT_PROCESS`]/[`with_current_process`]
//! below are exactly that -- `sched::resume::ResumeDriver::resume_for`/
//! `start` (the two places SCHED already knows the real `ProcessId` it is
//! about to resume) install it for the duration of the `resume()` call, and
//! [`real_nba_enqueue`] prefers it over the always-`0` raw ABI value. A test
//! exercising two concurrently-nonblocking-writing process instances in one
//! time step now exists (`vj-harness`'s
//! `two_process_nba_same_var_matches_iverilog_bit_for_bit`, against a real
//! Icarus run) -- see that fixture's own doc comment for why it uses a
//! causal (not same-event-race) dependency between the two processes: the
//! LRM leaves same-event multi-process activation order unspecified, and
//! `commit_nba`'s own hazard resolution (`sched::notify::commit_nba`)
//! already, deliberately, ties break by push/insertion order rather than by
//! `ProcessId` (see that function's own doc comment) -- this fix's real
//! contribution is that `source` is no longer silently dishonest about which
//! process staged a write, not a change to that already-correct resolution
//! algorithm.
//!
//! **r2 addendum (review fix, A2-part2 r2, finding 3):** the harness test
//! named above proves `commit_nba`'s push-order resolution, not that
//! `CURRENT_PROCESS` itself is wired correctly -- `commit_nba` never reads
//! `source` at all, so that differential test cannot distinguish "wired
//! correctly" from "absent/always-0". [`tests::
//! real_nba_enqueue_prefers_current_process_over_the_raw_abi_source`] closes
//! that gap directly: it calls [`real_nba_enqueue`] with and without
//! [`with_current_process`] installed and asserts the recorded `source`
//! differs accordingly, independent of `commit_nba`/push order entirely.
//!
//! ## Frame-storage deviation from the plan's literal 3-field struct
//!
//! The plan's own struct sketch is `{ kernel, activation: Box<ActivationRecord>,
//! resume_pc: u32 }`. `ActivationRecord::frame` is (correctly -- see that
//! field's own doc comment) only a non-owning [`crate::process::ArenaRange`]
//! *descriptor* into a separate, STATE-owned per-process frame arena; no such
//! frame-arena allocator exists anywhere in this tree yet (`state::instance::
//! StateEngine::allocate_instance` allocates the *signal* arena, a different
//! one, and no other committed code in this wave or an earlier one builds a
//! frame-word allocator). Since the frozen ABI's `local: *mut u8` parameter
//! must point at real, live, correctly-sized byte storage for `resume()` to
//! be callable at all, [`CompiledKernelProcess`] owns that byte buffer
//! directly (the `frame` field below) until a real frame-arena allocator
//! lands in a later wave and this adapter is updated to resolve
//! `activation.frame` against it instead.
use std::cell::{Cell, RefCell};
use std::ffi::CStr;

use vj_abi::eval::{EvalOutcome, EvalResult, RescheduleReq};
use vj_abi::{ProcessFn, SimHandle};

use super::reschedule_req_proposal::{
    KIND_BY_REQUEST, KIND_DELAY, KIND_DONE, KIND_EDGE, KIND_EDGE_LIST, KIND_FORK_JOIN,
    KIND_REPEAT_EDGE, KIND_WAIT_FORK, RescheduleReqLayoutProposal,
};
use super::{
    CompiledProcess, EventTerm, EventWaitSet, JoinKind, ProcessId, ProcessRequest, ProcessState,
    ResumeArg, SchedulerOps, SpawnSpec, Suspend,
};
use crate::dpi_task::{CallPoll, ExportRequest, TaskCall};
use crate::process::{ActivationRecord, ArenaRange, PredicateFn, WaitSpecStorage};
use crate::state::{EdgeKind, FourStateValue, InstanceHandle, StateCellId};

/// Byte offset, within [`RescheduleReq::opaque`], where this module stashes
/// the raw `next resume_pc` a caller must pass on its following call --
/// immediately after the 24-byte [`RescheduleReqLayoutProposal`] (PR-121's
/// "8 of the 32 reserved bytes free for a future field" headroom).
const NEXT_RESUME_PC_OFFSET: usize = 24;
/// Byte offset of the flags word in the `opaque` bytes' reserved tail.
const FLAGS_OFFSET: usize = 28;
/// Flag bit: the wait is a `wait(expr)` -- see [`EventWaitSet::rechecks_on_wake`].
const FLAG_RECHECKS_ON_WAKE: u8 = 1;
const PROPOSAL_SIZE: usize = 24;

/// The compile-time-resolved suspend shape [`build_eval_result`] encodes.
/// Deliberately **not** `vj_sim::process::cfg::SuspendPoint` (an
/// `sv_lang`-carrying, `frontend`-gated type): this module -- and the whole
/// `sched` module tree -- must build with `--no-default-features` (D4), so
/// every type crossing this boundary is `sv_lang`-free. `vj-codegen`
/// resolves its own `SuspendPoint` (a constant delay tick count, or an edge
/// signal's already-assigned `SlotId`) down to this shape before calling in.
#[derive(Clone, Copy, Debug)]
pub enum ResolvedSuspend {
    Delay(u64),
    /// A delay computed at run time (`#d`, `#(expr)`): the kernel stores
    /// the tick count with [`real_sv_rt_set_delay`] just before it returns.
    DynamicDelay,
    Edge {
        cell: StateCellId,
        edge: EdgeKind,
    },
    Done,
    /// **Task 6 (CG-081/CG-171):** a single-cell, single-edge REPEAT wait --
    /// `count` occurrences of `edge` on `cell` must fire before the process
    /// resumes. Closes `SuspendPoint::CycleDelay` (`##N`, always reachable
    /// via a `Timed` statement -- FULLY resolved by this task) and would
    /// close `RepeatedEventControl` (`repeat(n) @(...)`, CG-081's own name
    /// for this kind) too, were that AST shape ever reachable from a `Timed`
    /// statement at all -- see `vj-codegen::process_lower::
    /// resolve_suspend_point`'s `CycleDelay` arm and `vj-sim::process::
    /// cfg::SuspendPoint`'s own doc comment for the full reasoning on both.
    /// `count` is the RAW, un-multiplied cycle count -- `edge`s of `cell`
    /// are counted LIVE at scheduler runtime (`EventTerm::Repeat`,
    /// SC-052/SC-053, already built and tested), never pre-multiplied
    /// against any period at codegen time (ARCHITECTURE §6.5).
    RepeatEdge {
        count: u64,
        cell: StateCellId,
        edge: EdgeKind,
    },
    /// **Task 7 (AB-096/CG-082):** a single-cell, single-edge wait carrying a
    /// compiled re-check guard -- the `wait(expr)`/`iff`-qualified shape.
    /// See [`RescheduleReqLayoutProposal::from_single_edge_with_guard`]'s
    /// own doc comment for the full ABI-budget reasoning behind reusing
    /// `a`/`b` instead of a ninth struct field, and for why this is NOT
    /// extended to `RepeatEdge` in this pass.
    EdgeWithGuard {
        cell: StateCellId,
        edge: EdgeKind,
        guard_fn: extern "C" fn(SimHandle) -> bool,
    },
    /// **R-run Task 17 (PR-047/PR-061/PR-085):** a `fork ... join`
    /// (`JoinKind::All` only -- this task's own scope) suspend. Unlike
    /// every other variant above, a fork/join site's real payload (one
    /// [`super::SpawnSpec`] per branch) is NOT self-contained here -- it is
    /// compile-time-known (branches are static source, not a per-instance
    /// runtime count) but too large to fit `RescheduleReqLayoutProposal`'s
    /// frozen 24-byte budget for more than ~2 branches, so `vj-codegen`
    /// instead materializes a real, compile-time-constant
    /// `Vec<SpawnSpec>` fork table (one per compiled program, growing as
    /// each fork/join site is lowered) and this variant carries only
    /// `base`/`count`, an [`crate::process::ArenaRange`]-shaped reference
    /// into it -- exactly [`RescheduleReqLayoutProposal::from_fork_join`]'s
    /// own, already-specified `KIND_FORK_JOIN` wire shape (`docs/CONTRACTS.md`
    /// open question 8 / D1a's own "no allocator, no STATE-owned storage"
    /// concern does not apply here: this table is compiled INTO the
    /// program as an ordinary Rust `Vec` a `CompiledKernelProcess` is handed
    /// at construction time -- see [`decode_reschedule_with_fork_table`]).
    ForkJoin {
        join_kind: JoinKind,
        base: u32,
        count: u32,
    },
    /// `wait fork;` (SC-125) -- see `reschedule_req_proposal::KIND_WAIT_FORK`.
    WaitFork,
    /// `p.await()` / `p.suspend()` (classes plan Task 13) -- see
    /// `reschedule_req_proposal::KIND_BY_REQUEST`.
    ByRequest,
    /// A wait on several terms (`@*`, `@(a or posedge b)`, `wait(expr)` on a
    /// wide or multi-signal condition, `@(a)` any-change): the terms live in
    /// the program's wait table (`[base, base+count)`, see [`WaitTableEntry`]),
    /// encoded as `KIND_EDGE_LIST`.
    EventList {
        base: u32,
        count: u32,
    },
}

/// One term of a multi-term wait, in the program's wait table: the waited-on
/// signal's TEMPLATE slot (mapped to the waiting instance's cell at run time)
/// and its edge -- `None` = any value change of any bit.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct WaitTableEntry {
    pub slot: u32,
    pub edge: Option<EdgeKind>,
    /// `Some((lsb, width))`: the term watches only bits `[lsb, lsb + width)`
    /// of the signal -- a constant bit/part select (`@(posedge v[1])`, LRM
    /// 9.4.2) -- and an edge is taken on the select's least significant bit.
    pub select: Option<(u32, u32)>,
}

impl WaitTableEntry {
    /// A term on the whole signal.
    pub fn whole(slot: u32, edge: Option<EdgeKind>) -> Self {
        Self {
            slot,
            edge,
            select: None,
        }
    }
}

/// PR-121: builds a real `vj_abi::eval::EvalResult` from a
/// [`ResolvedSuspend`] and the raw `next_resume_pc` a caller must pass on its
/// next call for this same process (a plain `u32`, `vj-codegen`'s
/// `lower_kernel_fn` supplies the already-known target `SegId` here at
/// codegen time; see [`decode_reschedule`] for the inverse this module gives
/// the ABI's opaque bytes real meaning through, layered on top of
/// `RescheduleReqLayoutProposal`'s already-tested 24-byte tagged union rather
/// than inventing a second encoding).
pub fn build_eval_result(resolved: ResolvedSuspend, next_resume_pc: u32) -> EvalResult {
    let (outcome, proposal) = match resolved {
        ResolvedSuspend::Delay(ticks) => (
            EvalOutcome::SuspendedOnDelay,
            RescheduleReqLayoutProposal::from_delay(ticks),
        ),
        ResolvedSuspend::DynamicDelay => (
            EvalOutcome::SuspendedOnDelay,
            RescheduleReqLayoutProposal::from_delay(DYNAMIC_DELAY_MARK),
        ),
        ResolvedSuspend::Edge { cell, edge } => (
            EvalOutcome::SuspendedOnEvent,
            RescheduleReqLayoutProposal::from_single_edge(edge, cell),
        ),
        ResolvedSuspend::Done => (EvalOutcome::Completed, RescheduleReqLayoutProposal::done()),
        ResolvedSuspend::RepeatEdge { count, cell, edge } => (
            EvalOutcome::SuspendedOnEvent,
            RescheduleReqLayoutProposal::from_repeat_edge(count, edge, cell),
        ),
        // Task 7 (AB-096/CG-082).
        ResolvedSuspend::EdgeWithGuard {
            cell,
            edge,
            guard_fn,
        } => (
            EvalOutcome::SuspendedOnEvent,
            RescheduleReqLayoutProposal::from_single_edge_with_guard(edge, cell, guard_fn),
        ),
        // R-run Task 17: `EvalOutcome` is the frozen `vj-abi` 5-variant
        // enum (Completed/SuspendedOnEvent/SuspendedOnDelay/
        // MoreWorkInRegion/RuntimeError) -- no dedicated fork/join
        // discriminant exists in it (out of this task's authority to add
        // one, ARCHITECTURE §7 item 3), so this reuses `SuspendedOnEvent`,
        // the same "process suspended waiting on something" outcome
        // `KIND_EDGE`/`KIND_EDGE_LIST` above already use; `CompiledKernelProcess
        // ::resume` never reads `outcome` at all (only `decode_reschedule`'s
        // decode of `reschedule.opaque` matters for real dispatch), so this
        // choice is a documentation nicety, not load-bearing.
        ResolvedSuspend::ForkJoin {
            join_kind,
            base,
            count,
        } => (
            EvalOutcome::SuspendedOnEvent,
            RescheduleReqLayoutProposal::from_fork_join(join_kind, ArenaRange { base, count }),
        ),
        ResolvedSuspend::WaitFork => (
            EvalOutcome::SuspendedOnEvent,
            RescheduleReqLayoutProposal::wait_fork(),
        ),
        ResolvedSuspend::ByRequest => (
            EvalOutcome::SuspendedOnEvent,
            RescheduleReqLayoutProposal::by_request(),
        ),
        ResolvedSuspend::EventList { base, count } => (
            EvalOutcome::SuspendedOnEvent,
            RescheduleReqLayoutProposal::from_edge_list(ArenaRange { base, count }),
        ),
    };

    let mut opaque = [0u8; vj_abi::eval::RESCHEDULE_REQ_SIZE];
    // Task 11 (A2-part2; found, not assigned): this used to be an `unsafe {
    // core::mem::transmute(proposal) }` of the `#[repr(C)]`
    // `RescheduleReqLayoutProposal` struct directly to `[u8; PROPOSAL_SIZE]`
    // -- which reads that struct's own compiler-inserted alignment padding
    // (between `b: u32` and `edge_signal: u64`) as real bytes, and a plain
    // struct literal never pins padding to a known value, only its named
    // fields. `RescheduleReqLayoutProposal::to_bytes` (its own doc comment
    // has the full story, including why a `mem::zeroed()`-based struct
    // literal fix was tried first and does not work) writes all 24 bytes
    // itself instead, explicit `0` for the padding's own offsets -- no
    // `unsafe`, no transmute, no indeterminate byte survives into this
    // `EvalResult`, which `vj-codegen::process_lower` freezes directly
    // into a compile-time LLVM IR constant.
    let proposal_bytes: [u8; PROPOSAL_SIZE] = proposal.to_bytes();
    opaque[..PROPOSAL_SIZE].copy_from_slice(&proposal_bytes);
    opaque[NEXT_RESUME_PC_OFFSET..NEXT_RESUME_PC_OFFSET + 4]
        .copy_from_slice(&next_resume_pc.to_le_bytes());
    // opaque[28..32] is the flags word (see `mark_rechecks_on_wake`): explicit
    // zero here, never left as poison.

    EvalResult {
        outcome,
        reschedule: RescheduleReq { opaque },
    }
}

/// Marks `er` (a `wait(expr)` suspend) as one whose resumed kernel re-checks
/// its condition, so the scheduler may replay a wake parked while the process
/// was suspended ([`EventWaitSet::rechecks_on_wake`]). Only the codegen of a
/// level-sensitive wait calls this.
pub fn mark_rechecks_on_wake(mut er: EvalResult) -> EvalResult {
    er.reschedule.opaque[FLAGS_OFFSET] |= FLAG_RECHECKS_ON_WAKE;
    er
}

/// Whether [`mark_rechecks_on_wake`] was applied to `er`.
fn rechecks_on_wake(er: &EvalResult) -> bool {
    er.reschedule.opaque[FLAGS_OFFSET] & FLAG_RECHECKS_ON_WAKE != 0
}

/// Review fix (A2-part2 r2, binding rule 8): the typed failure
/// [`decode_reschedule`] returns instead of panicking for `KIND_EDGE_LIST`/
/// `KIND_FORK_JOIN`/any out-of-range kind byte.
///
/// **Why this stops at a typed error rather than a full decode into
/// `Suspend::WaitEvents`'s multi-term shape / `Suspend::ForkJoin`, for these
/// two kinds specifically:** `RescheduleReqLayoutProposal::from_edge_list`/
/// `from_fork_join`'s own doc comments are explicit that their payload is
/// only an [`ArenaRange`] *reference* into, respectively, "a static,
/// elaboration-time-built edge-term table" and "the engine's dynamic
/// fork-child table" -- never the concrete `StateCellId`s/`SpawnSpec`s
/// themselves. Per `docs/CONTRACTS.md` open question 8 and resolved decision
/// D1a, **neither table is materialized by any wave-A track**: no allocator
/// and no STATE-owned storage exists yet to resolve `base`/`count` against.
/// Fabricating a `Suspend::WaitEvents`/`Suspend::ForkJoin` from just a range
/// with no real backing data would be a silently-wrong `Suspend` (exactly
/// what this decoder's own prior doc comment already refused to do) --
/// strictly worse than an honest, typed "cannot resolve this yet" error.
/// This variant carries the range/`join_kind` bytes through unchanged, so
/// the future task that finally builds those tables has everything it needs
/// without inventing a second wire format.
#[derive(Clone, Copy, PartialEq, Eq, Debug)]
pub enum DecodeRescheduleError {
    /// `kind == KIND_EDGE_LIST`: a multi-term `Suspend::WaitEvents` whose
    /// term data lives in the (not-yet-materialized) edge-term table this
    /// `ArenaRange` would index into.
    UnresolvedEdgeList { base: u32, count: u32 },
    /// `kind == KIND_FORK_JOIN`: a `Suspend::ForkJoin` whose `SpawnSpec`s
    /// live in the (not-yet-materialized) dynamic fork-child table this
    /// `ArenaRange` would index into. `join_kind` is the raw
    /// `JoinKind`-discriminant byte (`RescheduleReqLayoutProposal::
    /// fork_join_kind`), carried through undecoded since decoding it further
    /// buys nothing without the range it pairs with also being resolvable.
    UnresolvedForkJoin {
        join_kind: u8,
        base: u32,
        count: u32,
    },
    /// Any byte outside `KIND_DELAY..=KIND_DONE` -- an ABI/version mismatch;
    /// `build_eval_result` (the only producer in this tree) never emits one.
    InvalidKind(u8),
}

/// The inverse of [`build_eval_result`]'s suspend-shape half: decodes
/// `r.reschedule`'s opaque bytes back into the scheduler-facing [`Suspend`].
/// Every kind byte [`RescheduleReqLayoutProposal`] can hold is named and
/// handled -- `KIND_EDGE_LIST`/`KIND_FORK_JOIN` and any out-of-range byte
/// return a typed [`DecodeRescheduleError`] (never a panic: review fix,
/// A2-part2 r2, binding rule 8) instead of a silently-wrong `Suspend`. See
/// [`DecodeRescheduleError`]'s own doc comment for why those two kinds stop
/// at a typed error rather than a full decode.
pub fn decode_reschedule(r: EvalResult) -> Result<Suspend, DecodeRescheduleError> {
    let mut proposal_bytes = [0u8; PROPOSAL_SIZE];
    proposal_bytes.copy_from_slice(&r.reschedule.opaque[..PROPOSAL_SIZE]);
    // SAFETY: inverse of `build_eval_result`'s transmute above -- every
    // `EvalResult` this function is ever called with in this tree was itself
    // produced by that function, so `proposal_bytes` is a valid
    // `RescheduleReqLayoutProposal` bit pattern (a plain, `#[repr(C)]`,
    // all-fields-`Copy` struct: no bit pattern of the right size is invalid).
    let proposal: RescheduleReqLayoutProposal = unsafe { core::mem::transmute(proposal_bytes) };
    match proposal.kind {
        KIND_DELAY => {
            let ticks = u64::from(proposal.a) | (u64::from(proposal.b) << 32);
            if ticks == DYNAMIC_DELAY_MARK {
                return Ok(Suspend::Delay(DYNAMIC_DELAY.with(Cell::get)));
            }
            Ok(Suspend::Delay(ticks))
        }
        // Task 7 (AB-096): `proposal.guard_fn()` is `Some` iff this
        // `KIND_EDGE` payload was built via `from_single_edge_with_guard`
        // (a real `wait(expr)`/`iff`-qualified edge) -- see
        // `sc068_decode_reschedule_can_never_produce_a_some_iff_edge_term`'s
        // own doc comment for the "documented flip condition" this arm now
        // satisfies for that specific shape, while a plain `ResolvedSuspend
        // ::Edge` (a == b == 0) still decodes `iff: None` exactly as before.
        KIND_EDGE => Ok(Suspend::WaitEvents(EventWaitSet {
            terms: vec![EventTerm::Edge {
                cell: StateCellId(proposal.edge_signal),
                edge: decode_edge_kind_u8(proposal.edge_kind),
                iff: proposal.guard_fn().map(PredicateFn),
            }],
            rechecks_on_wake: rechecks_on_wake(&r),
        })),
        KIND_DONE => Ok(Suspend::Done),
        KIND_WAIT_FORK => Ok(Suspend::WaitFork),
        KIND_BY_REQUEST => Ok(Suspend::ByRequest),
        // Task 6 (CG-081/CG-171): a single-cell, single-edge REPEAT wait --
        // decodes directly into the ALREADY-tested `EventTerm::Repeat`-over-
        // `Edge` machinery (SC-052/SC-053), unlike `KIND_EDGE_LIST`/
        // `KIND_FORK_JOIN` below: this kind's whole payload (`edge_kind`/
        // `edge_signal`/`a`/`b`) is self-contained inline, no out-of-line
        // table to resolve.
        KIND_REPEAT_EDGE => Ok(Suspend::WaitEvents(EventWaitSet {
            terms: vec![EventTerm::Repeat {
                count: u64::from(proposal.a) | (u64::from(proposal.b) << 32),
                inner: Box::new(EventTerm::Edge {
                    cell: StateCellId(proposal.edge_signal),
                    edge: decode_edge_kind_u8(proposal.edge_kind),
                    iff: None,
                }),
            }],
            ..Default::default()
        })),
        KIND_EDGE_LIST => Err(DecodeRescheduleError::UnresolvedEdgeList {
            base: proposal.a,
            count: proposal.b,
        }),
        KIND_FORK_JOIN => Err(DecodeRescheduleError::UnresolvedForkJoin {
            join_kind: proposal.fork_join_kind,
            base: proposal.a,
            count: proposal.b,
        }),
        other => Err(DecodeRescheduleError::InvalidKind(other)),
    }
}

/// R-run Task 17 (PR-047/PR-061/PR-085): the real, resolvable sibling of
/// [`decode_reschedule`] for a caller that DOES have a materialized fork
/// table to resolve `KIND_FORK_JOIN`'s `(base, count)` `ArenaRange` against
/// -- see [`ResolvedSuspend::ForkJoin`]'s own doc comment for why that table
/// is a plain, compile-time-constant `Vec<SpawnSpec>` (never a runtime
/// allocator), and [`CompiledKernelProcess`]'s own `fork_table` field for
/// where a real caller gets one. Every OTHER kind byte delegates verbatim to
/// [`decode_reschedule`] (never a second, independently-maintained copy of
/// its decode logic) -- this function only ever adds behavior for the one
/// kind that function still cannot resolve on its own.
///
/// `fork_table[base..base+count]` out of bounds (an internal-invariant
/// violation -- `vj-codegen` never constructs a `(base, count)` pair
/// that does not index its own, matching fork table) re-returns the
/// original typed [`DecodeRescheduleError::UnresolvedForkJoin`] rather than
/// panicking, so a caller that mismatches a kernel with the wrong fork
/// table gets a loud, typed failure, never an out-of-bounds panic.
pub fn decode_reschedule_with_fork_table(
    r: EvalResult,
    fork_table: &[SpawnSpec],
) -> Result<Suspend, DecodeRescheduleError> {
    decode_reschedule_with_tables(r, fork_table, &[])
}

/// [`decode_reschedule_with_fork_table`] plus the program's wait table, so a
/// `KIND_EDGE_LIST` (multi-term wait, [`ResolvedSuspend::EventList`]) also
/// resolves: each entry becomes an `Edge` term (edged) or an `ImplicitAny`
/// term (any change), cells still TEMPLATE slots (the kernel process maps
/// them to its instance).
pub fn decode_reschedule_with_tables(
    r: EvalResult,
    fork_table: &[SpawnSpec],
    wait_table: &[WaitTableEntry],
) -> Result<Suspend, DecodeRescheduleError> {
    match decode_reschedule(r) {
        Err(DecodeRescheduleError::UnresolvedEdgeList { base, count }) => {
            let start = base as usize;
            let end = start.saturating_add(count as usize);
            let entries = wait_table
                .get(start..end)
                .ok_or(DecodeRescheduleError::UnresolvedEdgeList { base, count })?;
            Ok(Suspend::WaitEvents(EventWaitSet {
                terms: entries
                    .iter()
                    .map(|e| match (e.select, e.edge) {
                        (Some((lsb, width)), edge) => EventTerm::Select {
                            cell: StateCellId(u64::from(e.slot)),
                            edge: edge.unwrap_or(EdgeKind::None),
                            lsb,
                            width,
                        },
                        (None, Some(edge)) => EventTerm::Edge {
                            cell: StateCellId(u64::from(e.slot)),
                            edge,
                            iff: None,
                        },
                        (None, None) => {
                            EventTerm::ImplicitAny(vec![StateCellId(u64::from(e.slot))])
                        }
                    })
                    .collect(),
                rechecks_on_wake: rechecks_on_wake(&r),
            }))
        }
        other => decode_fork_join(other, fork_table),
    }
}

fn decode_fork_join(
    decoded: Result<Suspend, DecodeRescheduleError>,
    fork_table: &[SpawnSpec],
) -> Result<Suspend, DecodeRescheduleError> {
    match decoded {
        Err(DecodeRescheduleError::UnresolvedForkJoin {
            join_kind,
            base,
            count,
        }) => {
            let start = base as usize;
            let end = start.saturating_add(count as usize);
            let spawns =
                fork_table
                    .get(start..end)
                    .ok_or(DecodeRescheduleError::UnresolvedForkJoin {
                        join_kind,
                        base,
                        count,
                    })?;
            Ok(Suspend::ForkJoin(
                decode_join_kind_u8(join_kind),
                spawns.to_vec(),
            ))
        }
        other => other,
    }
}

/// The inverse of `JoinKind as u8` (used by
/// [`RescheduleReqLayoutProposal::from_fork_join`]) -- an out-of-range byte
/// (an ABI/version mismatch this tree never actually produces, since
/// `JoinKind` has exactly 3 fieldless variants) falls back to `JoinKind::
/// None`, mirroring [`decode_edge_kind_u8`]'s own identical "out-of-range
/// falls back to the safest/most conservative variant" convention just
/// below, rather than panicking on a byte this function cannot itself prove
/// is unreachable.
fn decode_join_kind_u8(b: u8) -> JoinKind {
    if b == JoinKind::All as u8 {
        JoinKind::All
    } else if b == JoinKind::Any as u8 {
        JoinKind::Any
    } else {
        JoinKind::None
    }
}

fn decode_edge_kind_u8(b: u8) -> EdgeKind {
    if b == EdgeKind::Posedge as u8 {
        EdgeKind::Posedge
    } else if b == EdgeKind::Negedge as u8 {
        EdgeKind::Negedge
    } else if b == EdgeKind::BothEdges as u8 {
        EdgeKind::BothEdges
    } else {
        EdgeKind::None
    }
}

/// PR-100/PR-101/PR-102/PR-108/PR-109/PR-110: the ONE adapter bridging a
/// JIT'd/AOT'd `vj_abi::ProcessFn` into `sched::CompiledProcess` -- see
/// this module's own doc comment for why it also owns its instance's frame
/// bytes directly (a documented deviation from the plan's literal 3-field
/// sketch) and for the `source: 0` NBA scope limit.
///
/// **Task S7 finding (the `SimHandle` this struct's own `resume()` calls the
/// kernel with is `self.sim_handle`, NEVER `resume()`'s own `state`
/// parameter):** verified against the real `Engine` (`sched::control.rs`),
/// which passes `SimHandle(std::ptr::null_mut())` to every `CompiledProcess::
/// resume()` call unconditionally (`sched` is deliberately STATE-agnostic --
/// it dispatches by `ProcessId` against a generic `Box<dyn SignalAccess>`,
/// never a per-instance `SimHandle`) -- before this task, no committed code
/// anywhere in this tree ever drove `CompiledKernelProcess` through a real
/// `Engine`, so this null-handle-always behavior, while real and by design on
/// SCHED's side, had never actually reached a compiled kernel's own
/// `sv_get_scalar`/`sv_set_scalar` calls in practice (every existing test
/// calls `.resume()` directly with a handle it already controls). Since each
/// `CompiledKernelProcess` is permanently bound to exactly one process
/// INSTANCE (and therefore exactly one STATE `SimHandle`) for its whole
/// life, the fix is for this adapter to own and use that handle itself,
/// ignoring whatever `resume()`'s own `state` parameter happens to carry --
/// consistent with SCHED's own "generic over `SignalAccess`, no `SimHandle`
/// concept" design, not a workaround of it.
/// Set on the parent word of a [`ForkKernel::captures`] pair that copies the
/// other way: child frame word -> parent frame word, when the child is done
/// (a recursive task's `output`/`inout` formals).
pub const RETURN_COPY: u32 = 1 << 31;

/// The words a finished fork child hands its parent: (parent word, value).
type ReturnBox = std::rc::Rc<RefCell<Option<Vec<(u32, u64)>>>>;

pub struct CompiledKernelProcess {
    pub kernel: ProcessFn,
    pub activation: Box<ActivationRecord>,
    pub resume_pc: u32,
    frame: Box<[u8]>,
    sim_handle: SimHandle,
    /// R-run Task 17 (PR-085): the whole compiled program's fork-child
    /// table (see [`ResolvedSuspend::ForkJoin`]'s own doc comment) --
    /// empty (`Arc::from([])`, [`Self::new`]'s default) for every kernel
    /// that never lowers a fork/join site, in which case
    /// [`decode_reschedule_with_fork_table`] never actually indexes it (no
    /// `KIND_FORK_JOIN` byte is ever produced for such a kernel). Shared
    /// (`Arc`, never a per-instance clone of the whole table) because every
    /// `CompiledKernelProcess` instance minted from kernels of the SAME
    /// compiled program indexes the SAME table.
    fork_table: std::sync::Arc<[SpawnSpec]>,
    /// The whole compiled program's kernels, indexed by [`SpawnSpec::kernel`]
    /// (`None` for an index that is not a fork-branch kernel, e.g. a
    /// top-level process). Shared like `fork_table`; empty for a program
    /// with no fork/join site. [`CompiledProcess::mint_fork_child`] builds
    /// each branch's child process from it.
    fork_kernels: std::sync::Arc<[Option<ForkKernel>]>,
    /// Whether `sim_handle` is a real STATE `InstanceHandle`, so waited-on
    /// template slots can be mapped to this instance's cells (see `resume`).
    map_slots_to_instance: bool,
    /// The program's multi-term wait table ([`WaitTableEntry`]), shared like
    /// `fork_table`; empty when no kernel emits a multi-term wait.
    wait_table: std::sync::Arc<[WaitTableEntry]>,
    /// The instance and `(resume pc, label, exit pc)` table a `disable` of
    /// another process finds this one by ([`CompiledProcess::disable_exit`]).
    disable_instance: Option<u32>,
    disable_scopes: std::sync::Arc<[(u32, String, u32)]>,
    /// The last suspend was a fork/join wait ([`CompiledProcess::redirect_resume`]).
    joined: bool,
    /// The process ran to its end: nothing is left to disable.
    finished: bool,
    /// Every kernel resume runs on a fiber, so that C code of an import task
    /// it calls can block in an export task (see [`Self::with_task_fibers`]).
    task_fibers: bool,
    /// The assertion inputs the process's first run reads
    /// ([`proc_assert::set_action_inputs`]): the sampled arguments of a task
    /// spawned from a sequence match item (16.11); taken by that run.
    start_inputs: Option<Vec<i64>>,
    /// The kernel call parked on its fiber inside an import task (LRM 35.5.3),
    /// until the export task it asked for has run.
    call: Option<TaskCall>,
    /// Where the fiber leaves the `EvalResult` the kernel returned.
    fiber_result: std::rc::Rc<Cell<Option<EvalResult>>>,
    /// What the parked [`Self::call`] is waiting for; read by
    /// [`CompiledProcess::mint_fork_child`] to start the export task.
    request: Option<ExportRequest>,
    /// On an export-task child: the argument image of the import task's
    /// thunk, which receives the child's frame when the child is done so the
    /// `output`/`inout` formals reach the C caller.
    image_back: Option<(usize, usize)>,
    /// On a fork child: its frame words (child word, parent word) the parent
    /// gets when the child is done, and where it leaves them.
    returns: Vec<(u32, u32)>,
    return_box: Option<ReturnBox>,
    /// The children this process waits for that return words, copied into
    /// the frame when the process resumes.
    pending_returns: RefCell<Vec<ReturnBox>>,
}

/// One fork-branch kernel a [`CompiledKernelProcess`] can mint a child from:
/// its entry point and its own frame size (see [`CompiledKernelProcess::new`]).
#[derive(Clone, Debug)]
pub struct ForkKernel {
    pub kernel: ProcessFn,
    pub frame_word_count: u32,
    /// The variables the branch shares with the process that forks it:
    /// (parent frame word, child frame word) of each one's cell pointer,
    /// copied into the child as it is minted.
    pub captures: std::sync::Arc<[(u32, u32)]>,
    /// The branch's `(resume pc, label, exit pc)` disable scopes: a `disable`
    /// of a block nested in the branch from another process (LRM 9.6.2).
    pub disable_scopes: std::sync::Arc<[(u32, String, u32)]>,
}

// SAFETY: `sim_handle` is a `SimHandle` minted by `state::instance::
// InstanceHandle::into_sim_handle` and never dereferenced by this struct
// itself (only handed, opaquely, to the compiled kernel fn pointer, exactly
// like `resume()`'s own `state: SimHandle` parameter that every other
// `CompiledProcess` implementor already receives across whatever thread
// `Engine` runs on). `vj-sim`'s own `Engine`/`ProcessTable` already
// require `CompiledProcess: Send` for every registered process, so a
// `CompiledKernelProcess` that could never cross a thread boundary would be
// useless to register at all; this mirrors that same existing, accepted
// obligation (STATE's `InstanceHandle`/`Arena` are themselves designed to be
// accessed from whichever thread ENGINE's WOCB-barrier discipline currently
// grants exclusive access to, never two threads concurrently).
unsafe impl Send for CompiledKernelProcess {}

impl CompiledKernelProcess {
    /// `frame_word_count` is `vj-sim::process::liveness::FrameLayout`'s
    /// own total word count for this process TEMPLATE (each word 8 bytes,
    /// matching every other word-addressed frame convention in this
    /// workspace) -- `0` for a process with no declared locals at all (the
    /// fast-path, suspend-free case), in which case `resume()` passes a null
    /// `local` pointer, matching `vj-codegen::process_lower`'s own
    /// fast-path lowering, which never dereferences it. `sim_handle` is this
    /// process's own, permanently-bound `SimHandle` (see this struct's own
    /// doc comment for why `resume()` uses it, not its `state` parameter) --
    /// `SimHandle(std::ptr::null_mut())` is a legal value here too, and is
    /// exactly what every pre-Task-S7 test in this file already passes.
    pub fn new(kernel: ProcessFn, frame_word_count: u32, sim_handle: SimHandle) -> Self {
        let byte_len = frame_word_count as usize * 8;
        Self {
            kernel,
            activation: Box::new(ActivationRecord {
                pc: 0,
                frame: ArenaRange {
                    base: 0,
                    count: frame_word_count,
                },
                pending_wait: WaitSpecStorage::None,
                fork_state: None,
            }),
            resume_pc: 0,
            frame: vec![0u8; byte_len].into_boxed_slice(),
            sim_handle,
            fork_table: std::sync::Arc::from(Vec::new()),
            fork_kernels: std::sync::Arc::from(Vec::new()),
            map_slots_to_instance: false,
            wait_table: std::sync::Arc::from(Vec::new()),
            disable_instance: None,
            disable_scopes: std::sync::Arc::from(Vec::new()),
            joined: false,
            finished: false,
            task_fibers: false,
            start_inputs: None,
            call: None,
            fiber_result: std::rc::Rc::new(Cell::new(None)),
            request: None,
            image_back: None,
            returns: Vec::new(),
            return_box: None,
            pending_returns: RefCell::new(Vec::new()),
        }
    }

    /// Runs every resume of the kernel on a fiber (LRM 35.5.3, 35.7): when C
    /// code of an import task calls an export task which consumes time, the
    /// fiber parks with the kernel's whole stack, this process becomes the
    /// parent of the export task's process (`Suspend::ForkJoin`), and the
    /// kernel continues where it was once that process is done. Only for a
    /// design with a timed export task; any other pays nothing. The children
    /// this process forks inherit it.
    pub fn with_task_fibers(mut self) -> Self {
        self.task_fibers = true;
        self
    }

    /// Makes this process reachable by a `disable <label>;` of the other
    /// processes of `instance` (LRM 9.6.2): while it is suspended and will
    /// resume at `pc`, a disable of `label` resumes it at the exit `exit` of
    /// that block or task activation instead, for each `(pc, label, exit)`.
    pub fn with_disable_scopes(
        mut self,
        instance: u32,
        scopes: std::sync::Arc<[(u32, String, u32)]>,
    ) -> Self {
        self.disable_instance = Some(instance);
        self.disable_scopes = scopes;
        self
    }

    /// The values the process's first run reads as its assertion inputs.
    pub fn with_start_inputs(mut self, inputs: Vec<i64>) -> Self {
        self.start_inputs = Some(inputs);
        self
    }

    /// Installs the program's multi-term wait table (see [`WaitTableEntry`]).
    pub fn with_wait_table(mut self, wait_table: std::sync::Arc<[WaitTableEntry]>) -> Self {
        self.wait_table = wait_table;
        self
    }

    /// Declares `sim_handle` a real STATE `InstanceHandle` (what `vj
    /// run` always passes): the template slots a kernel waits on are then
    /// mapped to this instance's own cells. Must not be set for a handle
    /// that points at anything else (e.g. a unit test's mock state).
    pub fn with_instance_slot_mapping(mut self) -> Self {
        self.map_slots_to_instance = true;
        self
    }

    /// R-run Task 17 (PR-085): a builder-style setter installing this
    /// instance's own fork-child table -- see [`Self::fork_table`]'s own
    /// doc comment. A real caller (e.g. a fork-child minter closure built
    /// from `vj-codegen`'s own emitted table) calls this once, right
    /// after [`Self::new`], for every kernel drawn from a compiled program
    /// that contains at least one fork/join site (harmless to call for one
    /// that does not -- the table is simply never consulted).
    pub fn with_fork_table(mut self, fork_table: std::sync::Arc<[SpawnSpec]>) -> Self {
        self.fork_table = fork_table;
        self
    }

    /// Installs the program's fork-branch kernels (see [`Self::fork_kernels`]).
    pub fn with_fork_kernels(mut self, fork_kernels: std::sync::Arc<[Option<ForkKernel>]>) -> Self {
        self.fork_kernels = fork_kernels;
        self
    }

    /// Task 7 (PR-116, XC-044): a read-only view of this instance's own
    /// permanent frame bytes, exactly as `resume()` hands them to the
    /// compiled kernel's `local` parameter. Needed so a test can prove a
    /// declared local that the SV source never reads back (PR-017/PR-018's
    /// "every declared local gets a permanent slot regardless of use," the
    /// one claim no SV-visible read/write can observe on its own) still had
    /// its pre-suspend write survive, byte-for-byte, across a real
    /// suspend/resume round trip -- `activation`/`resume_pc` are already
    /// `pub` for the identical reason (direct internal-state inspection by
    /// a test, never by production scheduler code).
    pub fn frame_bytes(&self) -> &[u8] {
        &self.frame
    }
}

// PR-101/PR-102's ownership-boundary acceptance: this is the ONLY place a
// compiled kernel is bridged into a `CompiledProcess` -- see
// `exactly_one_compiledprocess_impl_over_a_kernel_backed_type` (this
// module's own test, below) for the structural proof and its documented
// carve-out for `sched::control::UnmintedForkChild` (a pre-existing,
// unrelated fork/join placeholder, not a kernel adapter).
impl CompiledProcess for CompiledKernelProcess {
    /// A fork branch runs in the parent's own instance (same `SimHandle`) and
    /// shares the program's fork tables, so a nested `fork` inside a branch
    /// mints its own children the same way.
    fn mint_fork_child(&self, spec: &SpawnSpec) -> Option<Box<dyn CompiledProcess>> {
        let fk = self.fork_kernels.get(spec.kernel as usize)?.clone()?;
        let mut child = CompiledKernelProcess::new(fk.kernel, fk.frame_word_count, self.sim_handle)
            .with_fork_table(self.fork_table.clone())
            .with_fork_kernels(self.fork_kernels.clone())
            .with_wait_table(self.wait_table.clone());
        child.map_slots_to_instance = self.map_slots_to_instance;
        child.task_fibers = self.task_fibers;
        // A timed export task, asked for by the import task parked in this
        // process: it runs in the instance of the C code's `svScope`, with the
        // thunk's argument image as its frame, which gets the frame back when
        // it is done.
        if let Some(req) = self.request
            && let Some(row) = self.fork_table.get(req.export as usize)
            && row.kernel == spec.kernel
            && row.entry_pc == spec.entry_pc
        {
            if !req.handle.is_null() {
                child.sim_handle = SimHandle(req.handle);
            }
            let n = child.frame.len();
            // SAFETY: `req.args` points at the image on the parked fiber's
            // stack, `frame_words` words (the thunk sized it from the same
            // kernel); it is alive while the request is outstanding.
            unsafe {
                std::ptr::copy_nonoverlapping(req.args, child.frame.as_mut_ptr(), n);
            }
            child.image_back = Some((req.args as usize, n));
        }
        // The branch runs in the parent's instance, so a `disable` of a block
        // nested in it finds the child like any other process there.
        child.disable_instance = self.disable_instance;
        child.disable_scopes = fk.disable_scopes.clone();
        // The variables it shares with this process: their cell pointers.
        for &(p, c) in fk.captures.iter() {
            if p & RETURN_COPY != 0 {
                child.returns.push((c, p & !RETURN_COPY));
                continue;
            }
            let (p, c) = (p as usize * 8, c as usize * 8);
            if p + 8 <= self.frame.len() && c + 8 <= child.frame.len() {
                child.frame[c..c + 8].copy_from_slice(&self.frame[p..p + 8]);
            }
        }
        if !child.returns.is_empty() {
            let bx: ReturnBox = std::rc::Rc::new(RefCell::new(None));
            child.return_box = Some(bx.clone());
            self.pending_returns.borrow_mut().push(bx);
        }
        Some(Box::new(child))
    }

    fn disable_instance(&self) -> Option<u32> {
        self.disable_instance
    }

    fn disable_exit(&self, label: &str) -> Option<u32> {
        if self.finished {
            return None;
        }
        if let Some(call) = &self.call {
            // LRM 35.9: `disable t;` of the import task this process is
            // blocked in (its C call is parked on an export task) ends the
            // call; the process resumes where it is, right after the call
            // returns ([`Self::redirect_resume`]).
            return (self.request.is_some() && call.import_name() == label)
                .then_some(self.resume_pc);
        }
        self.disable_scopes
            .iter()
            .find(|(pc, l, _)| *pc == self.resume_pc && l == label)
            .map(|(_, _, exit)| *exit)
    }

    fn owns_disable_scope(&self, label: &str) -> bool {
        self.disable_scopes.iter().any(|(_, l, _)| l == label)
    }

    fn redirect_resume(&mut self, pc: u32) -> bool {
        if let Some(call) = self.call.as_mut() {
            // The parked import call sees the disable (C: `svIsDisabledState`,
            // return 1) at its next poll; the export task it waits for is
            // killed with the rest of the joined children.
            call.disable();
            self.request = None;
            return std::mem::take(&mut self.joined);
        }
        self.resume_pc = pc;
        self.activation.pc = pc;
        std::mem::take(&mut self.joined)
    }

    fn resume(&mut self, _state: SimHandle, arg: ResumeArg, ops: &dyn SchedulerOps) -> Suspend {
        // Resuming from an event control or `wait`, or being disabled, is a
        // flush point for the process's pending deferred assertion reports
        // (16.4.2, `sched::deferred`); a delay is not (`#0` stays in the same
        // time step's Active/Inactive iteration, before any Observed region).
        if matches!(
            arg,
            ResumeArg::EventFired { .. } | ResumeArg::WaitOrderSatisfied | ResumeArg::Disabled
        ) && let Some(proc) = current_process()
        {
            crate::sched::deferred::flush_deferred(proc);
        }
        // ... and, when it resumes from an event control or `wait` (not from
        // a delay), for its pending procedural concurrent assertion instances
        // (16.14.6.2).
        if !matches!(arg, ResumeArg::Start | ResumeArg::TimeElapsed)
            && let Some(proc) = current_process()
        {
            crate::sched::proc_assert::flush(proc);
        }
        // Only an event control or `wait` resumption (or an `always_comb`
        // triggered again, which resumes the same way) is a violation report
        // flush point (12.4.2.1); a delay or a fork join is not.
        if matches!(
            arg,
            ResumeArg::EventFired { .. } | ResumeArg::WaitOrderSatisfied
        ) && let Some(proc) = current_process()
        {
            crate::sched::violation::flush_violation_reports(proc);
        }
        if let Some(inputs) = self.start_inputs.take() {
            crate::sched::proc_assert::set_action_inputs(inputs);
        }
        ABORTED.with(|c| c.set(false));
        let mask = current_process().map_or(0, take_fired);
        FIRED_MASK.with(|c| c.set(mask));
        FIRED_TERM.with(|c| {
            c.set(match arg {
                ResumeArg::EventFired { term_index } => term_index as i64,
                _ => -1,
            })
        });
        // The results the children joined since the last run returned.
        if !self.pending_returns.get_mut().is_empty() {
            let frame = &mut self.frame;
            self.pending_returns.get_mut().retain(|bx| {
                let Some(words) = bx.borrow_mut().take() else {
                    return true;
                };
                for (p, v) in words {
                    let at = p as usize * 8;
                    if at + 8 <= frame.len() {
                        frame[at..at + 8].copy_from_slice(&v.to_le_bytes());
                    }
                }
                false
            });
        }
        let local_ptr = if self.frame.is_empty() {
            std::ptr::null_mut()
        } else {
            self.frame.as_mut_ptr()
        };
        // Task S7 finding: `self.sim_handle`, never `resume()`'s own `state`
        // parameter -- see this struct's own doc comment.
        //
        // Task S7 (the end-to-end pipeline): `ops` is made reachable, for
        // exactly the duration of this raw kernel call, to `nba_enqueue`'s/
        // `sv_rt_finish`'s real production bodies below via `CURRENT_OPS` --
        // see that thread-local's own doc comment for why this is the one
        // sound hook point (the frozen `ProcessFn` ABI itself carries no
        // scheduler-ops parameter, and `SimHandle` reaches only STATE).
        let handle = self.sim_handle;
        let result: EvalResult = if self.task_fibers {
            if self.call.is_none() {
                let (kernel, pc, out) = (self.kernel, self.resume_pc, self.fiber_result.clone());
                let local = local_ptr as usize;
                self.call = TaskCall::new(move || {
                    out.set(Some(kernel(handle, local as *mut _, pc)));
                    0
                });
                assert!(
                    self.call.is_some(),
                    "no stack could be mapped for a process fiber"
                );
            }
            let call = self.call.as_mut().expect("created above");
            match with_scheduler_ops(ops, || call.poll()) {
                CallPoll::Finished(_) => {
                    self.call = None;
                    self.request = None;
                    self.fiber_result
                        .take()
                        .expect("a finished process fiber left no result")
                }
                CallPoll::Request(req) => {
                    // The C code of an import task wants an export task run:
                    // wait for it as the parent of its process.
                    let row = self.fork_table.get(req.export as usize).unwrap_or_else(|| {
                        panic!(
                            "export task {} has no fork-table row (internal)",
                            req.export
                        )
                    });
                    let spec = row.clone();
                    self.request = Some(req);
                    self.joined = true;
                    return Suspend::ForkJoin(JoinKind::All, vec![spec]);
                }
            }
        } else {
            with_scheduler_ops(ops, || (self.kernel)(handle, local_ptr, self.resume_pc))
        };

        let mut pc_bytes = [0u8; 4];
        pc_bytes.copy_from_slice(
            &result.reschedule.opaque[NEXT_RESUME_PC_OFFSET..NEXT_RESUME_PC_OFFSET + 4],
        );
        self.resume_pc = u32::from_le_bytes(pc_bytes);
        self.activation.pc = self.resume_pc;

        // `CompiledProcess::resume` (owned by the **sched** track,
        // `docs/CONTRACTS.md`'s `vj-sim::sched` ownership table) is
        // frozen to return `Suspend`, not `Result` -- changing that trait's
        // signature is out of this task's authority (integration rules:
        // "do not change a contract owned by another track except
        // minimally"). `decode_reschedule`'s `Err` arms are exactly the
        // kinds `build_eval_result` above (the only producer of the bytes
        // this call ever decodes) never emits -- see `DecodeRescheduleError`'s
        // own doc comment -- so this `unreachable!` documents a real,
        // checked invariant of THIS call site, not a masked bug; it is
        // `decode_reschedule` itself (usable independently, e.g. by a future
        // caller that constructs its own `EvalResult`) that must never
        // panic, and now never does.
        // R-run Task 17: routes through `decode_reschedule_with_fork_table`
        // (never the plain `decode_reschedule` directly any more) so a
        // `KIND_FORK_JOIN` byte -- now a real, producible encoding, see
        // `ResolvedSuspend::ForkJoin` -- resolves against this instance's
        // own `self.fork_table` instead of always failing with
        // `UnresolvedForkJoin`. For every kernel with no fork/join site,
        // `self.fork_table` is empty and this call behaves identically to
        // the old direct `decode_reschedule` call (that kind byte is simply
        // never produced for such a kernel).
        let suspend = decode_reschedule_with_tables(result, &self.fork_table, &self.wait_table)
            .unwrap_or_else(|e| {
                unreachable!(
                    "CompiledKernelProcess::resume: decode_reschedule_with_fork_table returned \
                     {e:?}, but build_eval_result (the only producer of RescheduleReq bytes in \
                     this tree) never constructs an EdgeList/invalid kind, and a real \
                     KIND_FORK_JOIN's (base,count) always indexes this same instance's own \
                     fork_table (an internal invariant of how vj-codegen pairs a kernel \
                     with its program's fork table -- see CompiledKernelProcess::with_fork_table)"
                )
            });
        self.joined = matches!(
            suspend,
            Suspend::ForkJoin(JoinKind::All | JoinKind::Any, _)
                | Suspend::WaitFork
                | Suspend::WaitOrder(_)
        );
        self.finished = matches!(suspend, Suspend::Done);
        if self.finished
            && let Some(bx) = self.return_box.take()
        {
            let words = self
                .returns
                .iter()
                .map(|&(c, p)| {
                    let at = c as usize * 8;
                    let mut w = [0u8; 8];
                    w.copy_from_slice(&self.frame[at..at + 8]);
                    (p, u64::from_le_bytes(w))
                })
                .collect();
            *bx.borrow_mut() = Some(words);
        }
        if self.finished
            && let Some((image, n)) = self.image_back.take()
        {
            // SAFETY: see `mint_fork_child`; the importing fiber is still
            // parked on that image.
            unsafe {
                std::ptr::copy_nonoverlapping(self.frame.as_ptr(), image as *mut u8, n);
            }
        }
        // Codegen encodes a waited-on signal by its TEMPLATE slot (the same
        // kernel serves every instance of its module); map it to THIS
        // instance's absolute cell here. Without this, every instance waited
        // on the first instance's signal (e.g. a second counter instance
        // clocked by its own `en` followed the first's `clk`). Only for a
        // handle that really is a STATE `InstanceHandle`
        // (`with_instance_slot_mapping`, set by `vj run`); unit-test
        // kernels run against mock state and keep their cells as given.
        if !self.map_slots_to_instance {
            return suspend;
        }
        match suspend {
            Suspend::WaitEvents(mut set) => {
                for term in &mut set.terms {
                    remap_term_to_instance(term, self.sim_handle);
                }
                Suspend::WaitEvents(set)
            }
            other => other,
        }
    }
}

/// See the remapping note in `CompiledKernelProcess::resume`.
fn remap_term_to_instance(term: &mut EventTerm, handle: SimHandle) {
    let map = |c: &mut StateCellId| {
        // The shared-variables and object-state cells are no instance's slot.
        if *c == OBJECT_STATE_CELL {
            OBJECT_STATE_WATCHED.store(true, std::sync::atomic::Ordering::Relaxed);
        } else if *c != SHARED_VARS_CELL {
            *c = crate::state::cell_for_slot(handle, vj_abi::SlotId(c.0 as u32));
        }
    };
    match term {
        EventTerm::Edge { cell, .. } | EventTerm::Select { cell, .. } => map(cell),
        EventTerm::ImplicitAny(cells) => cells.iter_mut().for_each(map),
        EventTerm::NamedEventTrigger(cell) => map(cell),
        EventTerm::Repeat { inner, .. } => remap_term_to_instance(inner, handle),
    }
}

/// The delay-ticks value marking a [`ResolvedSuspend::DynamicDelay`]: the
/// real value is the one the kernel last stored with
/// [`real_sv_rt_set_delay`] (the kernel call and its decode run back to back
/// on one thread).
const DYNAMIC_DELAY_MARK: u64 = u64::MAX;

thread_local! {
    static DYNAMIC_DELAY: Cell<u64> = const { Cell::new(0) };
}

/// `void sv_rt_set_delay(u64 ticks)`: the tick count of the run-time delay
/// the calling kernel is about to suspend on.
#[cfg_attr(feature = "aot-exports", unsafe(export_name = "sv_rt_set_delay"))]
pub extern "C" fn real_sv_rt_set_delay(ticks: u64) {
    DYNAMIC_DELAY.with(|c| c.set(ticks));
}

thread_local! {
    /// The index of the wait term that woke the running kernel (-1 if it
    /// was not woken by an event).
    static FIRED_TERM: Cell<i64> = const { Cell::new(-1) };
    /// Terms (bit k = term k, k < 64) that matched a process's wait since it
    /// last resumed, by process; moved to [`FIRED_MASK`] at resume.
    /// Indexed by process id (dense, from `register_process`): one load and
    /// store per matched term, no hashing, on the wake hot path.
    static FIRED_MASKS: std::cell::RefCell<Vec<u64>> = const { std::cell::RefCell::new(Vec::new()) };
    /// The resuming kernel's matched-term set (`sv_rt_fired_mask`).
    static FIRED_MASK: Cell<u64> = const { Cell::new(0) };
    /// Whether the running process ran `$finish`/`$stop`/`$fatal` (directly,
    /// in a subroutine, or as a null virtual interface's fatal error).
    static ABORTED: Cell<bool> = const { Cell::new(false) };
}

thread_local! {
    /// The cells holding variables a process shares with its fork branches
    /// (`sv_rt_cell_alloc`), kept for the whole run.
    static CELLS: std::cell::RefCell<Vec<Box<[u64]>>> = const { std::cell::RefCell::new(Vec::new()) };
}

/// `ptr sv_rt_cell_alloc(u64 bytes)`: a new zeroed cell for a variable a
/// process shares with its fork branches (LRM 1800-2023 §9.3.2: the forked
/// processes use the forking activation's own variable). One per creation
/// of such a variable, alive until the run ends.
#[cfg_attr(feature = "aot-exports", unsafe(export_name = "sv_rt_cell_alloc"))]
pub extern "C" fn real_sv_rt_cell_alloc(bytes: u64) -> *mut u8 {
    let words = (bytes as usize).div_ceil(8).max(1);
    let mut cell: Box<[u64]> = vec![0u64; words].into_boxed_slice();
    let ptr = cell.as_mut_ptr().cast::<u8>();
    CELLS.with(|c| c.borrow_mut().push(cell));
    ptr
}

/// The template slot a wait encodes for "a variable shared between
/// processes changed" (`SHARED_VARS_CELL`; never a real STATE slot).
pub const SHARED_VARS_SLOT: u32 = u32::MAX;

/// The cell a process waiting on a variable it shares with other processes
/// subscribes to: [`real_sv_rt_cell_store`] notifies it when such a variable
/// changes (the waiter's guard then re-checks its condition).
pub const SHARED_VARS_CELL: crate::state::StateCellId =
    crate::state::StateCellId(SHARED_VARS_SLOT as u64);

/// The template slot a wait encodes for "class-object state changed": a
/// `wait (...)` condition that reads a class property, calls a method of a
/// class, or calls `weak_reference::get` / `mailbox::num` reads heap state, and
/// a heap change is not a signal change. [`notify_object_state_change`] wakes
/// the waiters on it (the resume guard re-checks the condition). Never a real
/// STATE slot, like [`SHARED_VARS_SLOT`].
pub const OBJECT_STATE_SLOT: u32 = u32::MAX - 1;

/// The cell a wait on class-object state subscribes to (see
/// [`OBJECT_STATE_SLOT`]).
pub const OBJECT_STATE_CELL: crate::state::StateCellId =
    crate::state::StateCellId(OBJECT_STATE_SLOT as u64);

/// Whether any process has ever suspended on [`OBJECT_STATE_CELL`] in this
/// run. Heap writes are frequent; until a wait on object state exists the
/// notification is one relaxed load. Never cleared (a later wait may come back).
static OBJECT_STATE_WATCHED: std::sync::atomic::AtomicBool =
    std::sync::atomic::AtomicBool::new(false);

/// Marks object state as watched by a process that is statically sensitive to
/// [`OBJECT_STATE_CELL`] (a class covergroup's clocking-event sampler, LRM 19.4)
/// rather than suspended on it, so heap writes notify it (see
/// [`OBJECT_STATE_WATCHED`]).
pub fn watch_object_state() {
    OBJECT_STATE_WATCHED.store(true, std::sync::atomic::Ordering::Relaxed);
}

/// Whether any process has suspended on [`OBJECT_STATE_CELL`] (see
/// [`OBJECT_STATE_WATCHED`]); the scheduler's own notification sites (a process
/// ending is object state for `wait (p.status() == process::FINISHED)`) test it
/// before building a notify sink.
pub(crate) fn object_state_watched() -> bool {
    OBJECT_STATE_WATCHED.load(std::sync::atomic::Ordering::Relaxed)
}

/// A class object's observable state changed: a property was written, an
/// object was freed (a `weak_reference` to it reads null now), or a mailbox's
/// queue changed. Wakes the processes waiting on object state so they re-check
/// their conditions. Callers must not hold the objects/mailbox table lock.
/// Outside a running kernel (unit tests of the tables) there is nothing to wake.
pub fn notify_object_state_change() {
    if OBJECT_STATE_WATCHED.load(std::sync::atomic::Ordering::Relaxed) {
        with_current_ops(|ops| ops.notify_blocking_row_change(OBJECT_STATE_CELL));
    }
}

/// `void sv_rt_cell_store(ptr cell, ptr value, u64 bytes)`: stores a shared
/// variable's value into its cell, waking the processes waiting on shared
/// variables if it changed.
///
/// # Safety
/// `cell` and `value` each point to at least `bytes` valid bytes.
#[cfg_attr(feature = "aot-exports", unsafe(export_name = "sv_rt_cell_store"))]
pub unsafe extern "C" fn real_sv_rt_cell_store(cell: *mut u8, value: *const u8, bytes: u64) {
    // SAFETY: the caller's contract.
    let (to, from) = unsafe {
        (
            std::slice::from_raw_parts_mut(cell, bytes as usize),
            std::slice::from_raw_parts(value, bytes as usize),
        )
    };
    if to != from {
        to.copy_from_slice(from);
        with_current_ops(|ops| ops.notify_blocking_row_change(SHARED_VARS_CELL));
    }
}

thread_local! {
    /// The value each event-control term on a shared variable last saw:
    /// (waiting frame, site) -> its bits.
    static EDGE_SNAPSHOTS: std::cell::RefCell<std::collections::HashMap<(u64, u64), Vec<u8>>> =
        std::cell::RefCell::new(std::collections::HashMap::new());
}

/// [`ELEM_WAITS`]'s table: process -> `(site, array cell, row)` rows.
type ElemWaits = std::collections::HashMap<ProcessId, Vec<(u64, StateCellId, u64)>>;

thread_local! {
    /// Which element of which event array each process waits on
    /// (LRM 1800-2023 37.27 detail 1, `vpiWaitingProcesses` of an array
    /// element): process -> `(snapshot site, array cell, storage row)`,
    /// recorded by the element-term snapshots of its suspend. The scheduler
    /// wakes every waiter on the array's one cell; this is the only record
    /// of which element each one waits on. A snapshot of another suspend
    /// point (the site's high bits) replaces the process's older rows, and a
    /// reader intersects with the scheduler's waiters, so a process that has
    /// moved on is never reported.
    static ELEM_WAITS: std::cell::RefCell<ElemWaits> =
        std::cell::RefCell::new(std::collections::HashMap::new());
}

/// Bit of an event-array element term's snapshot site that marks the snapshot
/// as carrying `[row, valid, count, slot, handle]` (40 bytes) rather than a
/// bare value: see [`ELEM_WAITS`].
pub const ELEM_SITE_BIT: u64 = 1 << 62;

/// The processes now waiting on row `row` of the event array in `cell`
/// (a subset of the cell's waiters; see [`ELEM_WAITS`]).
pub fn event_elem_waiters(cell: StateCellId, row: u64) -> Vec<ProcessId> {
    ELEM_WAITS.with(|m| {
        let mut v: Vec<ProcessId> = m
            .borrow()
            .iter()
            .filter(|(_, rows)| rows.iter().any(|&(_, c, r)| c == cell && r == row))
            .map(|(p, _)| *p)
            .collect();
        v.sort_unstable();
        v
    })
}

/// The `EDGE_SNAPSHOTS` key of (`frame`, `site`). A process whose frame is
/// empty gets a null `frame` (`CompiledKernelProcess::resume`), which every
/// such process shares: it is keyed by the running process instead (the high
/// bit keeps that id apart from any real frame address).
fn snapshot_key(frame: u64, site: u64) -> (u64, u64) {
    match frame {
        0 => (
            current_process().map_or(0, |p| (1 << 63) | u64::from(p)),
            site,
        ),
        _ => (frame, site),
    }
}

/// `void sv_rt_edge_snapshot(u64 frame, u64 site, ptr value, u64 bytes)`:
/// records the value an event control on a shared variable waits to see
/// change (its process's `frame`, the term's `site`).
///
/// # Safety
/// `value` points to at least `bytes` valid bytes.
#[cfg_attr(feature = "aot-exports", unsafe(export_name = "sv_rt_edge_snapshot"))]
pub unsafe extern "C" fn real_sv_rt_edge_snapshot(
    frame: u64,
    site: u64,
    value: *const u8,
    bytes: u64,
) {
    // SAFETY: the caller's contract.
    let v = unsafe { std::slice::from_raw_parts(value, bytes as usize) }.to_vec();
    if site & ELEM_SITE_BIT != 0
        && bytes == 40
        && let Some(proc) = current_process()
    {
        let word = |i: usize| u64::from_le_bytes(v[i * 8..i * 8 + 8].try_into().unwrap());
        // Same suspend point (site bits above the term index): the wait
        // being re-entered replaces its own rows; another one starts over.
        let wait_of = |site: u64| site >> 16;
        ELEM_WAITS.with(|m| {
            let mut m = m.borrow_mut();
            let rows = m.entry(proc).or_default();
            rows.retain(|&(s, ..)| wait_of(s) == wait_of(site) && s != site);
            if word(1) != 0 {
                // SAFETY: word 4 is the process's own `SimHandle`, which the
                // snapshot's emitter loaded from the accessor context.
                let handle = SimHandle(word(4) as usize as *mut std::ffi::c_void);
                let cell = crate::state::cell_for_slot(handle, vj_abi::SlotId(word(3) as u32));
                rows.push((site, cell, word(0)));
            }
        });
    }
    EDGE_SNAPSHOTS.with(|m| m.borrow_mut().insert(snapshot_key(frame, site), v));
}

/// `i32 sv_rt_edge_check(u64 frame, u64 site, ptr value, u64 bytes, u32
/// width, u32 edge)`: whether the shared variable now holding `value` (a
/// packed 4-state value of `width` bits: values in bits 0.., unknown flags
/// from bit `width`) has made the event `edge` (0 any change, 1 posedge, 2
/// negedge, 3 either edge -- LRM 1800-2023 Table 9-2, on the least
/// significant bit) since the snapshot, which is then moved to `value`.
///
/// # Safety
/// `value` points to at least `bytes` valid bytes.
#[cfg_attr(feature = "aot-exports", unsafe(export_name = "sv_rt_edge_check"))]
pub unsafe extern "C" fn real_sv_rt_edge_check(
    frame: u64,
    site: u64,
    value: *const u8,
    bytes: u64,
    width: u32,
    edge: u32,
) -> i32 {
    // SAFETY: the caller's contract.
    let new = unsafe { std::slice::from_raw_parts(value, bytes as usize) }.to_vec();
    let bit = |buf: &[u8], i: usize| buf.get(i / 8).is_some_and(|b| b >> (i % 8) & 1 == 1);
    let used_bits = 2 * width as usize;
    let changed = |old: &[u8]| (0..used_bits).any(|i| bit(old, i) != bit(&new, i));
    let key = snapshot_key(frame, site);
    let fired = EDGE_SNAPSHOTS.with(|m| {
        let mut m = m.borrow_mut();
        let fired = match m.get(&key) {
            None => false,
            Some(old) => {
                // (known 0, known 1, unknown) of the least significant bit.
                let state = |buf: &[u8]| {
                    let (v, u) = (bit(buf, 0), bit(buf, width as usize));
                    (!u && !v, !u && v, u)
                };
                let (o0, o1, ox) = state(old);
                let (n0, n1, nx) = state(&new);
                let pos = (o0 && (n1 || nx)) || (ox && n1);
                let neg = (o1 && (n0 || nx)) || (ox && n0);
                match edge {
                    1 => pos,
                    2 => neg,
                    3 => pos || neg,
                    _ => changed(old),
                }
            }
        };
        m.insert(key, new.clone());
        fired
    });
    i32::from(fired)
}

/// `void sv_rt_edge_load(u64 frame, u64 site, ptr out, u64 bytes)`: copies
/// the bytes recorded by [`real_sv_rt_edge_snapshot`] for (`frame`, `site`)
/// into `out`; all zero when none was recorded (or it is shorter). Unlike
/// [`real_sv_rt_edge_check`] it leaves the snapshot in place, so a guard that
/// re-suspends keeps comparing against the state at the wait's start.
///
/// # Safety
/// `out` points to at least `bytes` writable bytes.
pub unsafe extern "C" fn real_sv_rt_edge_load(frame: u64, site: u64, out: *mut u8, bytes: u64) {
    // SAFETY: the caller's contract.
    let dst = unsafe { std::slice::from_raw_parts_mut(out, bytes as usize) };
    dst.fill(0);
    EDGE_SNAPSHOTS.with(|m| {
        if let Some(v) = m.borrow().get(&snapshot_key(frame, site)) {
            let n = v.len().min(dst.len());
            dst[..n].copy_from_slice(&v[..n]);
        }
    });
}

/// `void sv_rt_set_fired_term(i64 term)`: overrides which term of the wait
/// the resuming kernel reports from [`real_sv_rt_fired_term`]. A resume guard
/// of an event list holding array-element terms uses it: the scheduler wakes
/// all of them on the array's one cell and names the first, so the guard
/// replaces that with the term whose element actually triggered.
pub extern "C" fn real_sv_rt_set_fired_term(term: i64) {
    FIRED_TERM.with(|c| c.set(term));
}

/// Frees every cell of the previous run (`vj`'s run start).
pub fn reset_cells_for_new_run() {
    CELLS.with(|c| c.borrow_mut().clear());
    crate::state::reset_driver_counts_for_new_run();
    EDGE_SNAPSHOTS.with(|m| m.borrow_mut().clear());
    ELEM_WAITS.with(|m| m.borrow_mut().clear());
    // Process ids restart from zero: a waiter queued by the previous run must
    // not be mistaken for one of this run's processes.
    super::semaphore::reset_for_new_run();
    super::class_event::reset_for_new_run();
    super::mailbox::reset_for_new_run();
    crate::sched::proc_assert::reset();
    // 20.11: no assertion control task has run in a new run.
    crate::sched::assert_control::reset();
}

/// `i32 sv_rt_aborted()`: 1 if the running process has ended the run
/// (`$finish`/`$stop`/`$fatal`, LRM 1800-2023 §20.2/§20.10) since it
/// resumed -- code after such a call inside a subroutine must not run, and
/// neither may the caller's (`vj-codegen`'s `leave_if_aborted`).
#[cfg_attr(feature = "aot-exports", unsafe(export_name = "sv_rt_aborted"))]
pub extern "C" fn real_sv_rt_aborted() -> i32 {
    ABORTED.with(|c| i32::from(c.get()))
}

/// Records that term `term` of `proc`'s event wait matched in the current
/// time step (see [`real_sv_rt_fired_mask`]).
pub(crate) fn note_fired_term(proc: u32, term: usize) {
    if term < 64 {
        FIRED_MASKS.with(|m| {
            let mut m = m.borrow_mut();
            let i = proc as usize;
            if i >= m.len() {
                m.resize(i + 1, 0);
            }
            m[i] |= 1u64 << term;
        });
    }
}

/// Takes and clears `proc`'s recorded matched-term mask (0 if none).
fn take_fired(proc: u32) -> u64 {
    FIRED_MASKS.with(|m| {
        m.borrow_mut()
            .get_mut(proc as usize)
            .map_or(0, std::mem::take)
    })
}

/// Takes the terms (bit k = term k) of the calling process's event wait that
/// matched in the wake now running; for a native (non-kernel) process such as
/// a covergroup sampler, which nothing else drains. 0 outside a process.
pub fn take_fired_mask() -> u64 {
    current_process().map_or(0, take_fired)
}

/// `i64 sv_rt_fired_mask()`: every term of the event wait the calling kernel
/// resumed from that matched before it ran (bit k = term k), 0 if unknown.
/// Two members of one event list can fire in the same wake (LRM 1800-2023
/// 9.4.2.3) and `sv_rt_fired_term` names only the first.
#[cfg_attr(feature = "aot-exports", unsafe(export_name = "sv_rt_fired_mask"))]
pub extern "C" fn real_sv_rt_fired_mask() -> i64 {
    FIRED_MASK.with(|c| c.get()) as i64
}

/// `i64 sv_rt_fired_term()`: which term of the event wait the calling
/// kernel resumed from woke it (`wait_order`), -1 if none.
#[cfg_attr(feature = "aot-exports", unsafe(export_name = "sv_rt_fired_term"))]
pub extern "C" fn real_sv_rt_fired_term() -> i64 {
    FIRED_TERM.with(|c| c.get())
}

// --- Task S7: the `SimHandle`-reachability bridge for `nba_enqueue`/
// `sv_rt_finish`'s real production bodies. -----------------------------

thread_local! {
    /// While a compiled kernel's raw `ProcessFn` pointer is executing (inside
    /// [`with_scheduler_ops`], called only from [`CompiledKernelProcess::
    /// resume`] above), this holds the real `&dyn SchedulerOps` that
    /// `resume()` call was given -- the bridge this module's own doc comment
    /// (and `vj-codegen::process_lower::declare_rt_finish`'s doc comment)
    /// both flag as the still-open "the frozen ABI carries no scheduler-ops
    /// parameter, and `SimHandle` reaches only STATE" gap. This task closes
    /// it with the minimal, additive hook available: a thread-local set
    /// (and restored, not merely cleared, so a hypothetical nested call
    /// cannot leak its caller's `ops` early -- v1's single-threaded model
    /// never actually nests these calls, but the save/restore discipline
    /// costs nothing and removes the need to rely on that).
    ///
    /// `real_nba_enqueue`/`real_sv_rt_finish` below are its only readers;
    /// [`with_scheduler_ops`] is its only writer.
    ///
    /// Stored as a `'static`-erased trait-object pointer (a thread-local's
    /// contents must be `'static`, and `ops`'s real lifetime is only the
    /// duration of one `resume()` call) -- the `unsafe` lifetime-erasing
    /// transmute this requires lives in [`with_scheduler_ops`] alone, whose
    /// own save-and-restore discipline is exactly what keeps that erasure
    /// sound (the pointer is never read after the real `ops` it points at
    /// stops being alive).
    static CURRENT_OPS: RefCell<Option<*const (dyn SchedulerOps + 'static)>> =
        const { RefCell::new(None) };

    /// Review fix (A2-part2 r1, binding rule 7): the runtime-side lookup
    /// this module's own doc comment (and `process_lower::
    /// lower_nonblocking_assignment`'s doc comment) flagged as "no committed
    /// code implements yet". The frozen `ProcessFn` ABI carries no
    /// `ProcessId` parameter at all, so a compiled kernel itself can never
    /// pass its own identity to `nba_enqueue`; the *scheduler* (SCHED's own
    /// `ResumeDriver::resume_for`/`start`, `resume.rs`) already knows which
    /// `ProcessId` it is about to resume before it ever calls `.resume(..)`,
    /// so that is the one ambient-state hook this bridge needs -- mirrors
    /// [`CURRENT_OPS`] exactly (same save/restore discipline, same "set for
    /// the duration of one `resume()` call" scope). [`with_current_process`]
    /// is its only writer; [`real_nba_enqueue`] is its only reader.
    static CURRENT_PROCESS: RefCell<Option<ProcessId>> = const { RefCell::new(None) };

    /// Set while a kernel stages a clocking-block synchronous drive (between
    /// the `sv_rt_reactive_drive(1)` / `(0)` calls codegen brackets it with):
    /// the nonblocking writes it stages belong to the Re-NBA region (LRM
    /// 1800-2023 14.16), not the NBA region.
    static REACTIVE_DRIVE: std::cell::Cell<bool> = const { std::cell::Cell::new(false) };

    /// Within an edge-skewed clocking drive: the clock signal's cell, the edge
    /// (see [`EdgeWait`]) and the ticks after it (LRM 1800-2023 14.4).
    static EDGE_DRIVE: std::cell::Cell<Option<EdgeWait>> = const { std::cell::Cell::new(None) };
}

/// An edge output skew's wait (`output negedge #1 q`): the drive lands `after`
/// ticks past the next `edge` of `cell`.
#[derive(Clone, Copy)]
pub(super) struct EdgeWait {
    pub(super) cell: StateCellId,
    pub(super) edge: EdgeKind,
    pub(super) after: u64,
    /// A drive to the block's own event (counter `cell`, LRM 14.16): it
    /// matures now when the event was already triggered in this time step.
    pub(super) on_event: bool,
}

/// Whether the running kernel is inside a clocking-block synchronous drive.
pub(super) fn reactive_drive() -> bool {
    REACTIVE_DRIVE.with(std::cell::Cell::get)
}

/// The edge wait of the clocking drive the running kernel is staging, if its
/// output skew is an edge.
pub(super) fn edge_drive() -> Option<EdgeWait> {
    EDGE_DRIVE.with(std::cell::Cell::get)
}

/// The staging delay a kernel passes for a write that waits on
/// [`edge_drive`]'s edge instead of a tick count.
pub(super) const EDGE_DELAY: u64 = u64::MAX;

/// The real body of `sv_rt_reactive_drive`: `on != 0` opens, `0` closes the
/// span in which the kernel's nonblocking writes are a clocking drive staged
/// to the Re-NBA region (LRM 1800-2023 14.16). An open call with `edge != 0`
/// (1 posedge, 2 negedge, 3 either) is an edge output skew (14.4); `4` waits
/// for the next trigger of the clocking block's event (14.16; `clock_slot` is
/// its event counter): its writes
/// carry [`EDGE_DELAY`] and land `after` ticks past the next such edge of
/// `clock_slot`.
///
/// # Safety
/// `handle` must be the running kernel's instance handle.
#[cfg_attr(feature = "aot-exports", unsafe(export_name = "sv_rt_reactive_drive"))]
pub unsafe extern "C" fn real_sv_rt_reactive_drive(
    handle: SimHandle,
    on: u32,
    clock_slot: u32,
    edge: u32,
    after: u64,
) {
    if on != 0 {
        with_current_ops(|ops| ops.note_reactive_drive());
    }
    REACTIVE_DRIVE.with(|c| c.set(on != 0));
    let wait = (on != 0 && edge != 0).then(|| EdgeWait {
        cell: crate::state::cell_for_slot(handle, vj_abi::SlotId(clock_slot)),
        edge: match edge {
            1 => EdgeKind::Posedge,
            2 => EdgeKind::Negedge,
            _ => EdgeKind::BothEdges,
        },
        after,
        on_event: edge == 4,
    });
    EDGE_DRIVE.with(|c| c.set(wait));
}

/// Runs `f` (one `resume()` call, dispatched via `ResumeDriver`) with `proc`
/// reachable through [`CURRENT_PROCESS`] for exactly that duration,
/// restoring whatever was there before on return -- see that thread-local's
/// own doc comment. Called only from `sched::resume::ResumeDriver::
/// resume_for`/`start`, the two places `proc` is known before the dispatch
/// into `CompiledProcess::resume` happens. (`pub` so a runtime-ABI unit test
/// can stand in for a running process -- `vj_runtime::semaphore_rt`.)
pub fn with_current_process<R>(proc: ProcessId, f: impl FnOnce() -> R) -> R {
    let prev = CURRENT_PROCESS.with(|c| c.replace(Some(proc)));
    let result = f();
    CURRENT_PROCESS.with(|c| *c.borrow_mut() = prev);
    result
}

/// B-runtime Task 6 review fix r1 (RT-078): the read-only, cross-crate
/// half of the same [`CURRENT_PROCESS`] ambient-state bridge
/// [`real_nba_enqueue`] already uses to attribute an NBA write to a real
/// `ProcessId` despite the frozen `ProcessFn` ABI carrying none.
/// `vj_runtime::random_sink`'s `$random`/`$urandom`/`$urandom_range`
/// sink has the identical problem -- their ABI symbols
/// (`vj_codegen::expr::SV_RT_RANDOM_SYMBOL` et al.) are genuinely
/// zero-argument (no `SimHandle`, no `ProcessId`; see that module's own doc
/// comment for why), so this is the one already-live ambient hook that can
/// supply a real per-process identity to key a seed store by. Returns
/// `None` outside of any `ResumeDriver::resume_for`/`start` dispatch
/// (exactly [`with_current_process`]'s own "not currently installed" case) --
/// callers key an isolated fallback slot for that case, never conflating it
/// with a real `ProcessId`.
pub fn current_process() -> Option<ProcessId> {
    CURRENT_PROCESS.with(|c| *c.borrow())
}

/// Runs `f` (one raw kernel call) with `ops` reachable through
/// [`CURRENT_OPS`] for exactly that duration, restoring whatever was there
/// before on return (see that thread-local's own doc comment).
pub fn with_scheduler_ops<R>(ops: &dyn SchedulerOps, f: impl FnOnce() -> R) -> R {
    let raw: *const dyn SchedulerOps = ops;
    // SAFETY: erases `raw`'s real (sub-`'static`, tied to this call's own
    // `ops` parameter) lifetime so it can be stored in a `thread_local!`
    // (which requires `'static` contents). This is sound only because this
    // function itself restores `CURRENT_OPS` to `prev` (see below) before
    // returning -- i.e. before `ops`'s real borrow ends -- so no reader
    // (`with_current_ops`, called only from inside `f`'s own dynamic extent)
    // can ever observe this pointer after the object it points at is no
    // longer alive.
    let raw_static: *const (dyn SchedulerOps + 'static) = unsafe { std::mem::transmute(raw) };
    let prev = CURRENT_OPS.with(|c| c.replace(Some(raw_static)));
    let result = f();
    CURRENT_OPS.with(|c| *c.borrow_mut() = prev);
    result
}

/// Calls `f` with the current [`CURRENT_OPS`], if any is set. A miss (no
/// `with_scheduler_ops` call is currently on this thread's stack) is
/// unreachable for every kernel this tree's own codegen emits -- `nba_enqueue`/
/// `sv_rt_finish` are only ever invoked from inside a JIT'd kernel body, which
/// is only ever invoked from [`CompiledKernelProcess::resume`], which always
/// wraps the call in [`with_scheduler_ops`] -- so a miss here is silently
/// doing nothing for a call shape nothing in this tree can actually produce,
/// not a masked bug.
pub fn with_current_ops(f: impl FnOnce(&dyn SchedulerOps)) {
    // The pointer is copied out so the `RefCell` borrow ends before `f` runs:
    // `f` may itself enter `with_scheduler_ops` (an assertion action block
    // resumed from a Reactive callback, plan T29), which replaces the cell.
    let raw = CURRENT_OPS.with(|c| *c.borrow());
    if let Some(raw) = raw {
        // SAFETY: `raw` is only ever `Some` while the `resume()` call
        // that set it via `with_scheduler_ops` is still executing on
        // this thread (see that function's own body: it clears/restores
        // before returning), and that call's own `ops: &dyn
        // SchedulerOps` argument is alive for its whole duration -- so
        // this pointer is live for exactly as long as it can be read.
        let ops: &dyn SchedulerOps = unsafe { &*raw };
        f(ops);
    }
}

/// Task S7's real production body for the frozen `nba_enqueue` ABI symbol
/// (`vj_abi::process::NbaEnqueueFn`/`NBA_ENQUEUE_SYMBOL`) -- the ONE place
/// a nonblocking write a compiled kernel stages reaches the scheduler's real
/// [`SchedulerOps::stage_nba`]. Bound into a real JIT'd module via
/// `Jit::add_global_mapping(nba_fv, real_nba_enqueue as *const () as usize)`
/// (`vj::run`) -- no `#[no_mangle]`/linker-visible symbol is needed,
/// since `add_global_mapping` binds by explicit address, exactly like every
/// accessor-ABI stub this workspace's own tests already bind this same way.
#[cfg_attr(feature = "aot-exports", unsafe(export_name = "nba_enqueue"))]
pub extern "C" fn real_nba_enqueue(
    handle: SimHandle,
    slot: vj_abi::SlotId,
    value: u128,
    source: u32,
) {
    // SAFETY: identical justification to `state::accessor_impl::
    // sv_get_scalar` -- every real caller of this ABI-named symbol (the
    // JIT'd kernel `CompiledKernelProcess::resume` invokes) is handed a
    // `SimHandle` minted by `InstanceHandle::into_sim_handle`.
    let ih = unsafe { InstanceHandle::from_sim_handle(handle) };
    let cell = ih.cell_for_slot(slot.0 as usize);
    // Review fix (A2-part2 r1, binding rule 7): `source` (the frozen ABI's
    // own `u32` parameter) is always the codegen's constant `0` --
    // `process_lower::lower_nonblocking_assignment`'s own doc comment
    // explains why the frozen ABI has nowhere to carry a real `ProcessId`.
    // [`CURRENT_PROCESS`] is the real attribution: `ResumeDriver::
    // resume_for`/`start` (`sched::resume`) set it, via
    // [`with_current_process`], for the exact duration of the `resume()`
    // call that is (transitively, through this very kernel invocation)
    // calling this function right now, so a hit here is the ProcessId
    // actually being resumed. A miss (`None`) falls back to the raw ABI
    // value -- reachable only for the handful of this module's own unit
    // tests that call `CompiledKernelProcess::resume` directly, bypassing
    // `ResumeDriver` entirely; every real `vj run` dispatch goes
    // through `ResumeDriver`, so production code always hits `Some`.
    let source = CURRENT_PROCESS.with(|c| *c.borrow()).unwrap_or(source);
    with_current_ops(|ops| ops.stage_nba(cell, FourStateValue::Narrow(value), source));
}

/// A nonblocking write to part of a signal (`q[3:0] <= v`), to a memory row
/// (`mem[i] <= v`, `has_row != 0`), or to a >64-bit signal: stages
/// `(cell, row, value, mask)` for the NBA commit (`nba::MaskedNbaUpdate`).
/// `val`/`unk`/`mask` each point at `nwords` words holding the target's
/// `width` bits in codegen's val/unk planes, the new bits already shifted
/// into position; only the bits set in `mask` are written. A nonzero `delay`
/// (`q <= #d v`) defers the write to the time step `delay` ticks away.
///
/// # Safety
/// `val`, `unk` and `mask` must each point at `nwords` readable `u64`s, and
/// `nwords` must be `width.div_ceil(64)` (codegen's `build_nba_masked`).
#[cfg_attr(feature = "aot-exports", unsafe(export_name = "sv_rt_nba_masked"))]
pub unsafe extern "C" fn real_sv_rt_nba_masked(
    handle: SimHandle,
    slot: u32,
    row: u64,
    has_row: u32,
    val: *const u64,
    unk: *const u64,
    mask: *const u64,
    nwords: u32,
    width: u32,
    delay: u64,
) {
    // SAFETY: identical justification to `real_nba_enqueue` above.
    let ih = unsafe { InstanceHandle::from_sim_handle(handle) };
    let cell = ih.cell_for_slot(slot as usize);
    let n = nwords as usize;
    // SAFETY: the caller's contract (this function's `# Safety`).
    let (val, unk, mask) = unsafe {
        (
            std::slice::from_raw_parts(val, n).to_vec(),
            std::slice::from_raw_parts(unk, n).to_vec(),
            std::slice::from_raw_parts(mask, n).to_vec(),
        )
    };
    // STATE's encodings: `Narrow` is val / unk (unk at bit 64); `Wide` is
    // IEEE aval / bval, aval = val ^ unk.
    let value = if width <= 64 {
        FourStateValue::Narrow(u128::from(val[0]) | (u128::from(unk[0]) << 64))
    } else {
        FourStateValue::Wide {
            aval: val.iter().zip(&unk).map(|(v, u)| v ^ u).collect(),
            bval: unk,
        }
    };
    let row = (has_row != 0).then_some(row);
    let source = CURRENT_PROCESS.with(|c| *c.borrow()).unwrap_or(0);
    with_current_ops(|ops| {
        if delay == 0 {
            ops.stage_nba_masked(cell, row, value, mask, source);
        } else {
            ops.stage_nba_masked_delayed(cell, row, value, mask, source, delay);
        }
    });
}

/// The real body of `sv_rt_nba_coll`: a nonblocking write of element
/// `index` of the queue / dynamic array whose handle is held by `slot` (row
/// `row` when `has_row != 0`), LRM 1800-2023 10.4.2 / 7.10. `elem` points at
/// `elem_words` words (the element's packed bits); `flags` bit 0 is a queue
/// (an index equal to the size appends), bit 1 an object-handle element. The
/// handle is looked up and the index checked when the update commits, so an
/// out-of-range write is ignored then. `delay` ticks defer the staging.
/// A non-null `key` (`key_kind`/`key_width` as in `sv_rt_coll`) selects the
/// element of an associative array by key instead of `index`.
///
/// # Safety
/// `elem` must point at `elem_words` readable `u64`s, `key` (when non-null)
/// at the key's words.
#[cfg_attr(feature = "aot-exports", unsafe(export_name = "sv_rt_nba_coll"))]
pub unsafe extern "C" fn real_sv_rt_nba_coll(
    handle: SimHandle,
    slot: u32,
    row: u64,
    has_row: u32,
    index: i64,
    flags: u32,
    elem: *const u64,
    elem_words: u32,
    delay: u64,
    key: *const u64,
    key_kind: u32,
    key_width: u32,
) {
    // SAFETY: identical justification to `real_nba_enqueue` above.
    let ih = unsafe { InstanceHandle::from_sim_handle(handle) };
    let cell = ih.cell_for_slot(slot as usize);
    // SAFETY: the caller's contract (this function's `# Safety`).
    let elem = unsafe { std::slice::from_raw_parts(elem, elem_words as usize) }.to_vec();
    let obj = flags & 2 != 0;
    let mut shape = crate::collections::Shape {
        elem_words: elem_words as usize,
        assoc: false,
        key: crate::collections::KeyKind::Int {
            width: 0,
            signed: false,
        },
        obj,
        nested: crate::collections::Nested::NONE,
    };
    // An associative array's key is decoded now (a string key is a handle
    // to text that may change before the update lands).
    let mut key_val = None;
    if !key.is_null() {
        shape.assoc = true;
        shape.key = if key_kind == 2 {
            crate::collections::KeyKind::Str
        } else if key_kind == 3 {
            crate::collections::KeyKind::Obj
        } else {
            crate::collections::KeyKind::Int {
                width: key_width,
                signed: key_kind == 1,
            }
        };
        let kw = if key_kind == 2 {
            1
        } else {
            (2 * key_width as usize).div_ceil(64).max(1)
        };
        // SAFETY: the caller's contract: `key` points at the key's words.
        let bits = unsafe { std::slice::from_raw_parts(key, kw) };
        match shape.key_from_bits(bits) {
            Some(k) => key_val = Some(k),
            // An invalid key (x/z bits) writes nothing (LRM 7.8.6).
            None => {
                shape.discard(&elem);
                return;
            }
        }
    }
    let coll = super::nba::CollWrite {
        shape,
        index,
        key: key_val,
        queue: flags & 1 != 0,
    };
    let row = (has_row != 0).then_some(row);
    let source = CURRENT_PROCESS.with(|c| *c.borrow()).unwrap_or(0);
    with_current_ops(|ops| ops.stage_nba_coll(cell, row, elem, coll, source, delay));
}

/// Called by `sv_rt_obj`'s `FieldNba` (no symbol of its own): the real body of
/// a nonblocking class-property write: a nonblocking write of the `mask` bits
/// of the `words` packed words at `offset` of object `h`'s property storage
/// (`obj.p <= v`, `obj.v[3:0] <= v`; LRM 1800-2023 10.4.2, 8.x). The object is
/// retained until the NBA region commits it, so a handle overwritten or
/// dropped in between does not lose the update. A null or dead `h` stages
/// nothing. `delay` ticks defer the staging.
///
/// # Safety
/// `val` and `mask` must each point at `words` readable `u64`s.
pub unsafe extern "C" fn real_sv_rt_nba_field(
    h: u64,
    offset: u64,
    words: u32,
    val: *const u64,
    mask: *const u64,
    delay: u64,
) {
    if !crate::objects::is_live(h) {
        return;
    }
    let n = words as usize;
    // SAFETY: the caller's contract (this function's `# Safety`).
    let (val, mask) = unsafe {
        (
            std::slice::from_raw_parts(val, n).to_vec(),
            std::slice::from_raw_parts(mask, n).to_vec(),
        )
    };
    let w = super::nba::ObjWrite::new(h, offset as usize, val, mask);
    let source = CURRENT_PROCESS.with(|c| *c.borrow()).unwrap_or(0);
    with_current_ops(|ops| ops.stage_nba_obj(w, source, delay));
}

/// A nonblocking trigger (`->> e`, `->> #d e`; LRM 1800-2023 15.5.1) of class
/// event `id` (`sched::class_event`), `delay` ticks from now: staged like a
/// class-property write and committed in the NBA region, where it wakes the
/// event's waiters. A `null` id triggers nothing (15.5.5.2).
pub fn real_sv_rt_nba_class_event(id: u64, delay: u64) {
    if id == 0 {
        return;
    }
    let w = super::nba::ObjWrite::class_event(id);
    let source = CURRENT_PROCESS.with(|c| *c.borrow()).unwrap_or(0);
    with_current_ops(|ops| ops.stage_nba_obj(w, source, delay));
}

/// The real body of `sv_rt_nba_pulse`: a nonblocking event trigger
/// (`->> e`, `->> #d e`; LRM 1800-2023 15.5.2) of the event in `slot` (row
/// `row` of an event array when `has_row != 0`), `delay` ticks from now. The
/// trigger lands in the NBA region of its step (`NbaLog::stage_pulse`).
///
/// # Safety
/// `handle` must be a live `SimHandle` of the running instance.
#[cfg_attr(feature = "aot-exports", unsafe(export_name = "sv_rt_nba_pulse"))]
pub unsafe extern "C" fn real_sv_rt_nba_pulse(
    handle: SimHandle,
    slot: u32,
    row: u64,
    has_row: u32,
    delay: u64,
) {
    // SAFETY: identical justification to `real_nba_enqueue` above.
    let ih = unsafe { InstanceHandle::from_sim_handle(handle) };
    let cell = ih.cell_for_slot(slot as usize);
    let row = (has_row != 0).then_some(row);
    let source = CURRENT_PROCESS.with(|c| *c.borrow()).unwrap_or(0);
    with_current_ops(|ops| ops.stage_nba_pulse(cell, row, source, delay));
}

/// Task S7 finding: the real production body for the frozen `sv_set_scalar`
/// accessor-ABI symbol (`vj_abi::accessor::SV_SET_SCALAR_SYMBOL`), for a
/// COMPILED-KERNEL caller running under a real `Engine` -- deliberately NOT
/// `state::accessor_impl::sv_set_scalar` itself (that fn writes straight into
/// STATE's arena via `commit_write_at`'s own `&mut ()` no-op sink, which is
/// exactly correct for a debugger/VPI-style direct poke, but never notifies
/// `Engine`'s sensitivity table -- verified empirically while building this
/// task's own end-to-end pipeline: a compiled kernel's blocking write to a
/// clock signal through the bare accessor never woke a process waiting on its
/// posedge, because nothing told `Engine` the value had changed). A blocking
/// (`=`) write from a compiled process MUST go through
/// [`SchedulerOps::commit_blocking_write`] instead -- the same real path
/// `EngineOps::commit_blocking_write` (`sched::control`) already uses for
/// every OTHER blocking-write producer this crate has, which itself performs
/// the STATE write (through `Engine`'s own held `SignalAccess`) AND the
/// sensitivity-table notification together, as one atomic step. This
/// function therefore does NOT also call the bare `sv_set_scalar` -- doing so
/// would be a redundant second write through an unrelated path, not a
/// correctness requirement.
///
/// Bound into a real JIT'd module via `Jit::add_global_mapping(set_scalar_fv,
/// real_sv_set_scalar as *const () as usize)` (`vj::run`), in place
/// of the bare accessor -- the same binding mechanism [`real_nba_enqueue`]
/// above uses.
#[cfg_attr(feature = "aot-exports", unsafe(export_name = "sv_set_scalar"))]
pub extern "C" fn real_sv_set_scalar(handle: SimHandle, slot: vj_abi::SlotId, value: u128) {
    // SAFETY: identical justification to `real_nba_enqueue` above.
    let ih = unsafe { InstanceHandle::from_sim_handle(handle) };
    let cell = ih.cell_for_slot(slot.0 as usize);
    with_current_ops(|ops| ops.commit_blocking_write(cell, FourStateValue::Narrow(value)));
}

/// [`real_sv_set_scalar`]'s wide-tier sibling, bound in place of the bare
/// `state::sv_set_wide` for the identical reason: the bare accessor writes
/// STATE but never tells `Engine`, so a process waiting on a >64-bit signal
/// (`@w`, `wait (w == ...)`, a continuous assign or `always @*` reading it)
/// was never woken.
#[cfg_attr(feature = "aot-exports", unsafe(export_name = "sv_set_wide"))]
pub extern "C" fn real_sv_set_wide(
    handle: SimHandle,
    slot: vj_abi::SlotId,
    value: vj_abi::WideView,
) {
    // SAFETY: identical justification to `real_nba_enqueue` above.
    let ih = unsafe { InstanceHandle::from_sim_handle(handle) };
    let cell = ih.cell_for_slot(slot.0 as usize);
    let n = value.word_count as usize;
    // SAFETY: per the accessor-ABI contract, a `SvSetWideFn` caller's
    // `WideView` points at `word_count` initialized words in each plane for
    // the duration of the call (codegen's `build_set_wide` passes its own
    // stack buffers sized to the slot).
    let aval = unsafe { std::slice::from_raw_parts(value.aval, n) }.to_vec();
    let bval = unsafe { std::slice::from_raw_parts(value.bval, n) }.to_vec();
    with_current_ops(|ops| ops.commit_blocking_write(cell, FourStateValue::Wide { aval, bval }));
}

/// `sv_rt_huge`'s operation codes (`vj-codegen::huge`'s `HugeRt`).
pub const HUGE_OP_PRIME: u32 = 0;
pub const HUGE_OP_MARK: u32 = 1;
pub const HUGE_OP_COMMIT: u32 = 2;
pub const HUGE_OP_NBA: u32 = 3;

/// The one runtime entry for a compiled process's huge-signal shadow
/// (`state::huge_mirror`), `void* sv_rt_huge(u32 op, SimHandle, u32 slot,
/// u32 width, i64 a, i64 b, const u64* p, u64 delay)`:
///
/// * [`HUGE_OP_PRIME`]: the signal's long-lived PK image, current, at a stable
///   address (returned); `a != 0` re-reads the planes (external code wrote
///   them).
/// * [`HUGE_OP_MARK`]: widens the image's dirty window by the value bits
///   `[a, a + b)`.
/// * [`HUGE_OP_COMMIT`]: stores the dirty window to STATE and wakes the
///   processes sensitive to the signal iff its value changed.
///
/// * [`HUGE_OP_NBA`]: stages the nonblocking write `sig[a +: b] <= image`,
///   `p` the PK image of the `b`-bit right-hand side (val plane at bit 0, unk
///   plane at bit `b`), `delay` its intra-assignment delay; only the plane
///   words the select covers are staged.
///
/// `p` and `delay` are ignored by the other operations. Other operations
/// return null.
///
/// # Safety
/// For [`HUGE_OP_NBA`], `p` must point at `ceil(2 * b / 64)` readable words.
#[allow(clippy::too_many_arguments)]
#[cfg_attr(feature = "aot-exports", unsafe(export_name = "sv_rt_huge"))]
pub unsafe extern "C" fn real_sv_rt_huge(
    op: u32,
    handle: SimHandle,
    slot: u32,
    width: u32,
    a: i64,
    b: i64,
    p: *const u64,
    delay: u64,
) -> *mut u64 {
    match op {
        HUGE_OP_NBA => {
            if b > 0 && !p.is_null() {
                // SAFETY: the caller's contract (this function's `# Safety`).
                let img = unsafe { std::slice::from_raw_parts(p, (2 * b as usize).div_ceil(64)) };
                if let Some(w) =
                    crate::state::huge_nba_window(handle, vj_abi::SlotId(slot), width, a, b, img)
                {
                    let value = FourStateValue::Wide {
                        aval: w.aval,
                        bval: w.bval,
                    };
                    let source = CURRENT_PROCESS.with(|c| *c.borrow()).unwrap_or(0);
                    with_current_ops(|ops| {
                        ops.stage_nba_window(w.cell, w.window, value, w.mask, source, delay);
                    });
                }
            }
            std::ptr::null_mut()
        }
        HUGE_OP_PRIME => crate::state::huge_prime(handle, vj_abi::SlotId(slot), width, a != 0),
        HUGE_OP_MARK => {
            crate::state::huge_mark(handle, vj_abi::SlotId(slot), a, b);
            std::ptr::null_mut()
        }
        HUGE_OP_COMMIT => {
            huge_commit_and_notify(handle, slot, width);
            std::ptr::null_mut()
        }
        _ => std::ptr::null_mut(),
    }
}

/// Commits a huge signal's dirty window to STATE and wakes the processes
/// sensitive to it iff the value changed (`state::huge_commit`): the windowed
/// stand-in for `build_set_huge` + [`real_sv_set_wide`].
fn huge_commit_and_notify(handle: SimHandle, slot: u32, width: u32) {
    use crate::state::HugeCommit;
    let slot = vj_abi::SlotId(slot);
    let wants = |cell| {
        let answer = std::cell::Cell::new(false);
        with_current_ops(|ops| answer.set(ops.wants_delta(cell)));
        answer.get()
    };
    match crate::state::huge_commit(handle, slot, width, &wants) {
        HugeCommit::Nothing | HugeCommit::Unchanged => {}
        HugeCommit::Changed {
            cell,
            old,
            new,
            delta,
        } => {
            let edge = crate::sched::edge_between(old, new);
            with_current_ops(|ops| ops.notify_blocking_wide_change(cell, edge, delta));
        }
        HugeCommit::Full {
            cell,
            value,
            revalidate,
        } => {
            with_current_ops(|ops| ops.commit_blocking_write(cell, value));
            crate::state::huge_commit_full_done(handle, slot, revalidate);
        }
    }
}

/// [`real_sv_set_scalar`]'s memory-row sibling: commits the row through
/// STATE's own row path, then wakes processes sensitive to the memory iff
/// the row changed (`SchedulerOps::notify_blocking_row_change`).
#[cfg_attr(feature = "aot-exports", unsafe(export_name = "sv_set_mem_word"))]
pub extern "C" fn real_sv_set_mem_word(
    handle: SimHandle,
    slot: vj_abi::SlotId,
    row_index: u64,
    value: vj_abi::WideView,
) {
    let (handle, slot) = crate::state::resolve_foreign_slot(handle, slot);
    let (cell, changed) =
        crate::state::set_mem_word_reporting_change(handle, slot, row_index, value);
    if changed {
        with_current_ops(|ops| ops.notify_blocking_row_change(cell));
    }
}

/// Task S7's real production body for the frozen `sv_rt_finish` ABI symbol
/// (`vj-codegen::process_lower::SV_RT_FINISH_SYMBOL`/
/// `declare_rt_finish`) -- reaches the scheduler's real
/// [`SchedulerOps::request_finish`], the one canonical path
/// `Engine::run_until` already checks once per timestep (SC-075). Bound the
/// same way as [`real_nba_enqueue`] above.
#[cfg_attr(feature = "aot-exports", unsafe(export_name = "sv_rt_finish"))]
pub extern "C" fn real_sv_rt_finish(_handle: SimHandle, code: i32) {
    ABORTED.with(|c| c.set(true));
    with_current_ops(|ops| ops.request_finish(code));
}

/// `disable fork;` (LRM 9.6.3, SC-125): the real production body for
/// `sv_rt_disable_fork`. Kills every live descendant the CURRENT process
/// forked, through [`SchedulerOps::disable_fork`] -- reached via the same
/// [`CURRENT_OPS`]/[`CURRENT_PROCESS`] bridges as [`real_sv_rt_finish`]/
/// [`real_nba_enqueue`] (the frozen `ProcessFn` ABI carries no process id).
#[cfg_attr(feature = "aot-exports", unsafe(export_name = "sv_rt_disable_fork"))]
pub extern "C" fn real_sv_rt_disable_fork(_handle: SimHandle) {
    if let Some(proc) = current_process() {
        with_current_ops(|ops| ops.disable_fork(proc));
    }
}

/// `disable f;` in a branch of fork `f`: the real production body for
/// `sv_rt_disable_fork_block` ([`SchedulerOps::disable_fork_block`]).
#[cfg_attr(
    feature = "aot-exports",
    unsafe(export_name = "sv_rt_disable_fork_block")
)]
pub extern "C" fn real_sv_rt_disable_fork_block(_handle: SimHandle) {
    if let Some(proc) = current_process() {
        with_current_ops(|ops| ops.disable_fork_block(proc));
    }
}

/// A cross-process/fork-sibling `disable <label>;` (LRM 9.6.2): the real
/// production body for `sv_rt_disable_named` ([`SchedulerOps::
/// disable_named`]). `label` is a NUL-terminated C string (same convention
/// as `vj::dump::sv_rt_dump_vars`'s own `scopes` argument) -- a
/// static string literal `vj-codegen::process_lower` builds from the
/// disable statement's own source text, never attacker/user-controlled
/// binary data, so a `CStr`/UTF-8 decode failure here is an internal
/// codegen bug, surfaced loudly (`expect`), never swallowed. Sets this
/// kernel's own abort flag ([`abort_current_kernel`]) when the call ends
/// the CALLER itself (it is, or descends from, one of the named live
/// processes) -- `vj-codegen`'s caller then calls `leave_if_aborted`
/// right after, the same two-step `process::kill()` already uses.
///
/// # Safety
/// `label` must point to a valid, NUL-terminated, UTF-8 C string for the
/// duration of this call -- guaranteed by codegen always passing a `'static`
/// global string constant it compiled from the source `disable` statement.
#[cfg_attr(feature = "aot-exports", unsafe(export_name = "sv_rt_disable_named"))]
pub unsafe extern "C" fn real_sv_rt_disable_named(
    _handle: SimHandle,
    label: *const std::ffi::c_char,
) {
    // SAFETY: the caller's contract (this function's own doc comment).
    let label = unsafe { CStr::from_ptr(label) }
        .to_str()
        .expect("sv_rt_disable_named: codegen must pass a valid UTF-8 C string");
    if let Some(proc) = current_process() {
        let mut ends_caller = false;
        with_current_ops(|ops| ends_caller = ops.disable_named(proc, label));
        if ends_caller {
            abort_current_kernel();
        }
    }
}

/// Classes plan Task 13 (LRM 9.7): the `process` class's scheduler bridge,
/// called by `vj_runtime::process_class`. Like `real_sv_rt_disable_fork`
/// it reaches the scheduler through [`CURRENT_OPS`]/[`CURRENT_PROCESS`]
/// (the frozen `ProcessFn` ABI carries no process id). All four are no-ops
/// (or report `Running`) outside a kernel call.
///
/// Queues `req` for the calling process; `true` iff it terminates the
/// caller itself (the caller then leaves its kernel at once -- see
/// [`abort_current_kernel`]).
pub fn process_class_request(req: ProcessRequest) -> bool {
    let Some(requester) = current_process() else {
        return false;
    };
    let mut ends_caller = false;
    with_current_ops(|ops| ends_caller = ops.process_request(requester, req));
    ends_caller
}

/// `p.status()` for process `target`.
pub fn process_class_state(target: ProcessId) -> ProcessState {
    let mut state = ProcessState::Running;
    with_current_ops(|ops| state = ops.process_state(target));
    state
}

/// Marks the calling process as one a `process` handle exists for and
/// returns its id (`process::self()`), or `None` outside a process.
pub fn process_class_track_current() -> Option<ProcessId> {
    let proc = current_process()?;
    with_current_ops(|ops| ops.process_track(proc));
    Some(proc)
}

/// Makes the running kernel's next `leave_if_aborted` check fire, which ends
/// the calling process's segment (`Suspend::Done`) -- the one mechanism
/// `$finish` uses, reused here for a process that killed itself.
pub fn abort_current_kernel() {
    ABORTED.with(|c| c.set(true));
}

/// A `width`-bit window at bit `lo` of a `total`-bit cell: the value (from
/// codegen's `val`/`unk` lanes, `n` words each, LSB aligned) in the cell's
/// own shape, and the window's mask words.
///
/// # Safety
/// `val`/`unk` address `width.div_ceil(64)` words each.
unsafe fn force_window(
    total: u32,
    wide: bool,
    lo: u32,
    width: u32,
    val: *const u64,
    unk: *const u64,
) -> (FourStateValue, Vec<u64>) {
    let n = width.div_ceil(64).max(1) as usize;
    // SAFETY: the caller's contract.
    let (val, unk) = unsafe {
        (
            std::slice::from_raw_parts(val, n),
            std::slice::from_raw_parts(unk, n),
        )
    };
    let words = total.div_ceil(64).max(1) as usize;
    let (mut v, mut u, mut m) = (vec![0u64; words], vec![0u64; words], vec![0u64; words]);
    for i in 0..width {
        let (src, dst) = (i as usize, (lo + i) as usize);
        if dst >= total as usize {
            break;
        }
        let bit = |w: &[u64]| (w[src / 64] >> (src % 64)) & 1;
        v[dst / 64] |= bit(val) << (dst % 64);
        u[dst / 64] |= bit(unk) << (dst % 64);
        m[dst / 64] |= 1 << (dst % 64);
    }
    let value = if wide {
        FourStateValue::Wide {
            aval: v.iter().zip(&u).map(|(a, b)| a ^ b).collect(),
            bval: u,
        }
    } else {
        FourStateValue::Narrow(u128::from(v[0]) | (u128::from(u[0]) << 64))
    };
    (value, m)
}

/// [`crate::state::slot_state`], reporting (and aborting the running process
/// on) a slot `force` cannot reach.
fn slot_state_or_abort(
    handle: SimHandle,
    slot: vj_abi::SlotId,
) -> Option<(StateCellId, u32, bool, FourStateValue, u32)> {
    match crate::state::slot_state(handle, slot) {
        Ok(state) => Some(state),
        Err(what) => {
            // A refused force must not look like a run that went on: end the
            // simulation with a failing exit code (as `$finish(1)` does).
            eprintln!("vj: runtime error: {what} is not supported");
            real_sv_rt_finish(handle, 1);
            None
        }
    }
}

/// A flag bit of `sv_rt_force`'s `is_net` and `sv_rt_release`'s `flags`: the
/// statement is a procedural `assign`/`deassign`, not a `force`/`release`.
pub const FORCE_PROCEDURAL_ASSIGN: u32 = 2;

/// A flag bit of `sv_rt_force`'s `is_net`: a `force`'s continuous driver
/// re-applying it after an operand changed (LRM 10.6.2), not the statement
/// executing -- no force event.
pub const FORCE_REFRESH: u32 = 4;

/// A force, release, procedural assign or deassign, as reported to an
/// embedder that asked for them ([`record_force_events`]): VPI's
/// `cbForce`/`cbRelease`/`cbAssign`/`cbDeassign` (IEEE 1800-2023 38.36.1).
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct ForceEvent {
    /// The forced object's STATE cell.
    pub cell: StateCellId,
    pub release: bool,
    /// `assign`/`deassign` rather than `force`/`release`.
    pub procedural: bool,
}

thread_local! {
    /// `Some` while an embedder records force events.
    static FORCE_EVENTS: RefCell<Option<Vec<ForceEvent>>> = const { RefCell::new(None) };
}

/// Starts (or stops) recording force events for [`take_force_events`].
pub fn record_force_events(on: bool) {
    FORCE_EVENTS.with(|e| *e.borrow_mut() = on.then(Vec::new));
}

/// The force events since the last call, in order.
pub fn take_force_events() -> Vec<ForceEvent> {
    FORCE_EVENTS.with(|e| {
        e.borrow_mut()
            .as_mut()
            .map(std::mem::take)
            .unwrap_or_default()
    })
}

fn note_force(ev: ForceEvent) {
    FORCE_EVENTS.with(|e| {
        if let Some(v) = e.borrow_mut().as_mut() {
            v.push(ev);
        }
    });
}

/// `force lhs = v;` (LRM 1800-2023 §10.6.2): forces the `width`-bit window
/// at bit `lo` of `slot` to `v` (codegen's `val`/`unk` lanes) on behalf of
/// the `force` statement at source offset `site` (`crate::state::force`).
/// `flags`: bit 0 for a net (released, it goes back to its drivers), plus
/// [`FORCE_PROCEDURAL_ASSIGN`] and [`FORCE_REFRESH`].
///
/// # Safety
/// `val`/`unk` address `width.div_ceil(64)` words each.
#[allow(clippy::too_many_arguments)]
#[cfg_attr(feature = "aot-exports", unsafe(export_name = "sv_rt_force"))]
pub unsafe extern "C" fn real_sv_rt_force(
    handle: SimHandle,
    slot: vj_abi::SlotId,
    lo: u32,
    width: u32,
    val: *const u64,
    unk: *const u64,
    flags: u32,
    site: u64,
) {
    if let Some((row, lo_base)) = crate::state::slot_row_alias(handle, slot) {
        // A ref port bound to an element of an unpacked array (LRM
        // 23.3.3.2): the force is the element's. A ref port binds a
        // variable, never a net, so the net bit the formal's own flags may
        // carry does not apply (a released variable keeps its value).
        // SAFETY: the caller's contract.
        unsafe {
            force_row_impl(
                handle,
                slot,
                row,
                lo + lo_base,
                width,
                val,
                unk,
                flags & !1,
                site,
            );
        }
        return;
    }
    let Some((cell, total, wide, current, lo_base)) = slot_state_or_abort(handle, slot) else {
        return;
    };
    let lo = lo + lo_base;
    // SAFETY: the caller's contract.
    let (value, mask) = unsafe { force_window(total, wide, lo, width, val, unk) };
    let site = (handle.0 as u64, site);
    if flags & FORCE_REFRESH == 0 {
        note_force(ForceEvent {
            cell,
            release: false,
            procedural: flags & FORCE_PROCEDURAL_ASSIGN != 0,
        });
    }
    let write = crate::state::with_force_table(handle, |t| {
        let w = t.force(
            cell,
            value,
            mask,
            flags & 1 != 0,
            site,
            current,
            flags & FORCE_PROCEDURAL_ASSIGN != 0,
        );
        t.set_bypass(true);
        w
    });
    with_current_ops(|ops| ops.commit_blocking_write(cell, write));
    crate::state::with_force_table(handle, |t| t.set_bypass(false));
}

/// `release lhs;`: ends the force of the `width`-bit window at bit `lo` of
/// `slot`; a net goes back to its driven value. `flags` is
/// [`FORCE_PROCEDURAL_ASSIGN`] for a `deassign`, 0 for a `release`.
#[cfg_attr(feature = "aot-exports", unsafe(export_name = "sv_rt_release"))]
pub extern "C" fn real_sv_rt_release(
    handle: SimHandle,
    slot: vj_abi::SlotId,
    lo: u32,
    width: u32,
    flags: u32,
) {
    if let Some((row, lo_base)) = crate::state::slot_row_alias(handle, slot) {
        release_row_impl(handle, slot, row, lo + lo_base, width, flags);
        return;
    }
    let Some((cell, total, wide, _, lo_base)) = slot_state_or_abort(handle, slot) else {
        return;
    };
    let lo = lo + lo_base;
    note_force(ForceEvent {
        cell,
        release: true,
        procedural: flags & FORCE_PROCEDURAL_ASSIGN != 0,
    });
    let ones = vec![u64::MAX; width.div_ceil(64).max(1) as usize];
    // SAFETY: `ones` has the window's word count, used for both lanes.
    let (_, mask) = unsafe { force_window(total, wide, lo, width, ones.as_ptr(), ones.as_ptr()) };
    let procedural = flags & FORCE_PROCEDURAL_ASSIGN != 0;
    if let Some(restore) =
        crate::state::with_force_table(handle, |t| t.release(cell, &mask, procedural))
    {
        // The restored value is not a new driver write: with other windows
        // still forced it must not overwrite the entry's `driven` bits with
        // their forced values (LRM 10.6.2), so it bypasses the filter.
        crate::state::with_force_table(handle, |t| t.set_bypass(true));
        with_current_ops(|ops| ops.commit_blocking_write(cell, restore));
        crate::state::with_force_table(handle, |t| t.set_bypass(false));
    }
}

/// `force`/`assign` of row `row` of the unpacked array in `slot` (LRM
/// 1800-2023 10.6.2): forces the `width`-bit window at bit `lo` of that
/// element. [`real_sv_rt_force`] for one memory row; `flags` as there.
///
/// # Safety
/// `val`/`unk` address `width.div_ceil(64)` words each.
#[allow(clippy::too_many_arguments)]
unsafe fn force_row_impl(
    handle: SimHandle,
    slot: vj_abi::SlotId,
    row: u64,
    lo: u32,
    width: u32,
    val: *const u64,
    unk: *const u64,
    flags: u32,
    site: u64,
) {
    let (cell, at, total, current) = crate::state::row_state(handle, slot, row);
    // SAFETY: the caller's contract; a memory row is always wide.
    let (value, mask) = unsafe { force_window(total, true, lo, width, val, unk) };
    let procedural = flags & FORCE_PROCEDURAL_ASSIGN != 0;
    if flags & FORCE_REFRESH == 0 {
        note_force(ForceEvent {
            cell,
            release: false,
            procedural,
        });
    }
    let site = (handle.0 as u64, site);
    let write = crate::state::with_force_table(handle, |t| {
        let w = t.force_row(
            cell,
            at,
            value,
            mask,
            flags & 1 != 0,
            site,
            current,
            procedural,
        );
        t.set_bypass(true);
        w
    });
    commit_row(handle, slot, row, write);
    crate::state::with_force_table(handle, |t| t.set_bypass(false));
}

/// `release`/`deassign` of the `width`-bit window at bit `lo` of row `row`
/// of `slot`: [`real_sv_rt_release`] for one memory row.
fn release_row_impl(
    handle: SimHandle,
    slot: vj_abi::SlotId,
    row: u64,
    lo: u32,
    width: u32,
    flags: u32,
) {
    let (cell, at, total, _) = crate::state::row_state(handle, slot, row);
    let procedural = flags & FORCE_PROCEDURAL_ASSIGN != 0;
    note_force(ForceEvent {
        cell,
        release: true,
        procedural,
    });
    let ones = vec![u64::MAX; width.div_ceil(64).max(1) as usize];
    // SAFETY: `ones` has the window's word count, used for both lanes.
    let (_, mask) = unsafe { force_window(total, true, lo, width, ones.as_ptr(), ones.as_ptr()) };
    if let Some(restore) =
        crate::state::with_force_table(handle, |t| t.release_row(cell, at, &mask, procedural))
    {
        commit_row(handle, slot, row, restore);
    }
}

/// VPI `vpi_put_value(vpiForceFlag)` on row `row` of the unpacked array in
/// `slot` (IEEE 1800-2023 38.34, 10.6.2): [`force_row_impl`], `is_net` the
/// net flag.
///
/// # Safety
/// `val`/`unk` address `width.div_ceil(64)` words each.
#[allow(clippy::too_many_arguments)]
pub unsafe fn vpi_force_row(
    handle: SimHandle,
    slot: vj_abi::SlotId,
    row: u64,
    lo: u32,
    width: u32,
    val: *const u64,
    unk: *const u64,
    is_net: bool,
    site: u64,
) {
    // SAFETY: the caller's contract.
    unsafe {
        force_row_impl(
            handle,
            slot,
            row,
            lo,
            width,
            val,
            unk,
            u32::from(is_net),
            site,
        );
    }
}

/// VPI `vpi_put_value(vpiReleaseFlag)` on a row: [`release_row_impl`].
pub fn vpi_release_row(handle: SimHandle, slot: vj_abi::SlotId, row: u64, lo: u32, width: u32) {
    release_row_impl(handle, slot, row, lo, width, 0);
}

/// A flag bit of `sv_rt_force_row`'s `flags`: the call is a release (a
/// deassign with [`FORCE_PROCEDURAL_ASSIGN`]); `val`/`unk` are unused.
pub const FORCE_ROW_RELEASE: u32 = 8;

/// `force mem[i] = v;` / `release mem[i];` (and `assign`/`deassign`) on row
/// `row` of the unpacked array in `slot` (LRM 1800-2023 10.6.2):
/// [`real_sv_rt_force`]/[`real_sv_rt_release`] for one memory row, the
/// release when `flags` has [`FORCE_ROW_RELEASE`].
///
/// # Safety
/// `val`/`unk` address `width.div_ceil(64)` words each (unused for a
/// release).
#[allow(clippy::too_many_arguments)]
#[cfg_attr(feature = "aot-exports", unsafe(export_name = "sv_rt_force_row"))]
pub unsafe extern "C" fn real_sv_rt_force_row(
    handle: SimHandle,
    slot: vj_abi::SlotId,
    row: u64,
    lo: u32,
    width: u32,
    val: *const u64,
    unk: *const u64,
    flags: u32,
    site: u64,
) {
    if flags & FORCE_ROW_RELEASE != 0 {
        release_row_impl(handle, slot, row, lo, width, flags);
    } else {
        // SAFETY: the caller's contract.
        unsafe { force_row_impl(handle, slot, row, lo, width, val, unk, flags, site) };
    }
}

/// Stores `v` (a wide value) into row `row` of `slot` through the one write
/// primitive and wakes what is sensitive to the memory.
fn commit_row(handle: SimHandle, slot: vj_abi::SlotId, row: u64, v: FourStateValue) {
    let FourStateValue::Wide { mut aval, mut bval } = v else {
        return;
    };
    let view = vj_abi::accessor::WideView {
        aval: aval.as_mut_ptr(),
        bval: bval.as_mut_ptr(),
        word_count: aval.len() as u32,
    };
    let (handle, slot) = crate::state::resolve_foreign_slot(handle, slot);
    let (cell, changed) = crate::state::set_mem_word_reporting_change(handle, slot, row, view);
    if changed {
        with_current_ops(|ops| ops.notify_blocking_row_change(cell));
    }
}

/// Whether the `force` statement at source offset `site` of `handle`'s
/// instance is still in effect: its continuous driver re-forces only while
/// it is.
#[cfg_attr(feature = "aot-exports", unsafe(export_name = "sv_rt_force_active"))]
pub extern "C" fn real_sv_rt_force_active(handle: SimHandle, site: u64) -> u32 {
    u32::from(crate::state::with_force_table(handle, |t| {
        t.site_active((handle.0 as u64, site))
    }))
}

/// `$countdrivers` (IEEE 1800-2023 D.2, Table D.1) of bit `bit` of the net in
/// `slot`, whose strength key is `key`: writes `out[0..5]` = net_is_forced,
/// number_of_01x_drivers, number_of_0_drivers, number_of_1_drivers,
/// number_of_x_drivers. A net a resolver drives has its recorded counts; any
/// other net has at most one driver, so its value (the one its driver last
/// wrote, when forced) says whether that driver is at 0, 1, x or `z`.
///
/// # Safety
///
/// `out` points to five writable `u32`.
#[cfg_attr(feature = "aot-exports", unsafe(export_name = "sv_rt_net_drivers"))]
pub unsafe extern "C" fn real_sv_rt_net_drivers(
    handle: SimHandle,
    slot: vj_abi::SlotId,
    key: u64,
    bit: u32,
    out: *mut u32,
) {
    // Every exit leaves `out` defined: the aborting paths below return with the
    // run stopped, but the calling process still reads the five words, and an
    // unwritten stack slot made the JIT and AOT builds print different values.
    for i in 0..5 {
        // SAFETY: the caller's contract.
        unsafe { out.add(i).write(0) };
    }
    let Some((cell, _total, _wide, current, lo_base)) = slot_state_or_abort(handle, slot) else {
        return;
    };
    let (forced, driven) =
        crate::state::with_force_table(handle, |t| (t.contains(cell), t.driven_value(cell)));
    let counts = crate::state::driver_counts_of(key, bit).unwrap_or_else(|| {
        let v = driven.as_ref().unwrap_or(&current);
        match crate::state::four_state_bit_at(v, bit + lo_base) {
            vj_abi::Bit4::Zero => [1, 0, 0],
            vj_abi::Bit4::One => [0, 1, 0],
            vj_abi::Bit4::X => [0, 0, 1],
            vj_abi::Bit4::Z => [0, 0, 0],
        }
    });
    if counts == crate::state::UNCOUNTED {
        eprintln!(
            "vj: runtime error: $countdrivers of a net in a bidirectional switch network is not supported"
        );
        real_sv_rt_finish(handle, 1);
        return;
    }
    let all = [
        u32::from(forced),
        counts[0] + counts[1] + counts[2],
        counts[0],
        counts[1],
        counts[2],
    ];
    for (i, v) in all.into_iter().enumerate() {
        // SAFETY: the caller's contract.
        unsafe { out.add(i).write(v) };
    }
}

/// A `$strobe`/`$monitor` argument thunk codegen emits per call site:
/// re-reads the call's arguments from STATE and prints them (LRM 21.2.2/
/// 21.2.3 evaluate those arguments at the END of the timestep, not at the
/// call).
pub type PostponedThunk = extern "C" fn(SimHandle);

struct StrobeCallback {
    thunk: PostponedThunk,
    handle: SimHandle,
}

// SAFETY: `handle` is the kernel's own instance `SimHandle`, only ever
// passed back to the thunk on the scheduler's own thread (same obligation
// `CompiledKernelProcess`'s `unsafe impl Send` documents).
unsafe impl Send for StrobeCallback {}

impl super::RegionCallback for StrobeCallback {
    fn invoke(&mut self, _ctx: &dyn super::SchedulerCtx) {
        (self.thunk)(self.handle);
    }
}

/// `$strobe(...)`: run `thunk` once, in this timestep's Postponed region.
#[cfg_attr(
    feature = "aot-exports",
    unsafe(export_name = "sv_rt_strobe_postponed")
)]
pub extern "C" fn real_sv_rt_strobe(handle: SimHandle, thunk: PostponedThunk) {
    with_current_ops(|ops| {
        ops.register_region_callback(
            vj_abi::Region::Postponed,
            ops.now(),
            Box::new(StrobeCallback { thunk, handle }),
        );
    });
}

thread_local! {
    /// The active `$monitor`'s generation (0 = none yet). Only one monitor
    /// is ever active (LRM 21.2.3): starting a new one bumps this, which
    /// retires the old one's callback. Thread-local: one simulation runs on
    /// one thread.
    static MONITOR_GENERATION: std::cell::Cell<u64> = const { std::cell::Cell::new(0) };
    /// `$monitoron`/`$monitoroff`.
    static MONITOR_ENABLED: std::cell::Cell<bool> = const { std::cell::Cell::new(true) };
}

/// High bit of `sv_rt_monitor_start`'s slot count: the slot array carries one
/// extra trailing entry, the mask of `$time`-like display values, and the
/// monitor prints only when some other value differs from the last print
/// (LRM 21.2.3: "each time a variable or an expression in the argument list
/// changes value").
pub const MONITOR_DEDUPE_N: u32 = 1 << 31;

/// What one monitor-thunk call is doing (see [`MonitorDedupe`]).
#[derive(Clone, Copy, PartialEq, Eq)]
enum MonitorMode {
    /// Not inside a thunk call.
    Idle,
    /// Evaluating the arguments at a watched-cell update: the call compares
    /// them with the last evaluated values, sets `due` and prints nothing.
    Probe,
    /// The Postponed print: records the printed values and prints.
    Print,
}

/// The active plain `$monitor`'s change-test state (LRM 21.2.3: a display is
/// due when some argument EXPRESSION took a different value at some update
/// during the step, even if a later update restored it): the mask of
/// `$time`-like display values, the values (as an opaque key) last evaluated,
/// and the per-step "due" flag.
struct MonitorDedupe {
    mask: u32,
    mode: MonitorMode,
    last: Option<Vec<u8>>,
    /// A display is owed at the end of this step (set by a probe that saw a
    /// changed expression, or by `$monitoron`; cleared by the print).
    due: bool,
    /// The thunk and instance handle a watched-cell update re-evaluates.
    probe: Option<(PostponedThunk, SimHandle)>,
}

thread_local! {
    static MONITOR_DEDUPE: std::cell::RefCell<Option<MonitorDedupe>> =
        const { std::cell::RefCell::new(None) };
}

/// A new `$monitor` starts: `Some(mask)` turns the expression-value change
/// test on (bit `i` of `mask` exempts display value `i`), `None` off. The
/// first display is always due.
pub fn monitor_dedupe_reset(mask: Option<u32>) {
    MONITOR_DEDUPE.with(|d| {
        *d.borrow_mut() = mask.map(|mask| MonitorDedupe {
            mask,
            mode: MonitorMode::Idle,
            last: None,
            due: true,
            probe: None,
        });
    });
}

fn monitor_dedupe_set_mode(mode: MonitorMode) {
    MONITOR_DEDUPE.with(|d| {
        if let Some(m) = d.borrow_mut().as_mut() {
            m.mode = mode;
        }
    });
}

/// Arms (or disarms) the print-side change record around one monitor thunk
/// call from the Postponed callback.
pub fn monitor_dedupe_arm(on: bool) {
    monitor_dedupe_set_mode(if on {
        MonitorMode::Print
    } else {
        MonitorMode::Idle
    });
}

/// A watched cell changed value: when the active monitor tests expression
/// values, evaluates its arguments now (display suppressed) and marks a
/// display due if they differ from the last evaluation.
pub fn monitor_probe() {
    let probe = MONITOR_DEDUPE.with(|d| {
        d.borrow()
            .as_ref()
            .filter(|m| m.mode == MonitorMode::Idle)
            .and_then(|m| m.probe)
    });
    if let Some((thunk, handle)) = probe {
        monitor_dedupe_set_mode(MonitorMode::Probe);
        thunk(handle);
        monitor_dedupe_set_mode(MonitorMode::Idle);
    }
}

/// `true` while the active `$monitor` tests expression values.
fn monitor_dedupe_active() -> bool {
    MONITOR_DEDUPE.with(|d| d.borrow().is_some())
}

/// Reads and clears the active monitor's per-step "display due" flag.
fn monitor_dedupe_take_due() -> bool {
    MONITOR_DEDUPE.with(|d| {
        d.borrow_mut()
            .as_mut()
            .is_some_and(|m| std::mem::take(&mut m.due))
    })
}

/// Builds this call's value key (`make_key` gets the `$time`-like mask) and
/// applies the mode: `Probe` compares it with the last evaluation, remembers
/// it, marks the display due on a difference, and returns `true` (suppress the
/// print); `Print` remembers it and returns `false`. Always `false` otherwise.
pub fn monitor_dedupe_check(make_key: impl FnOnce(u32) -> Vec<u8>) -> bool {
    MONITOR_DEDUPE.with(|d| {
        let mut d = d.borrow_mut();
        let Some(m) = d.as_mut().filter(|m| m.mode != MonitorMode::Idle) else {
            return false;
        };
        let key = make_key(m.mask);
        let probing = m.mode == MonitorMode::Probe;
        if probing && m.last.as_deref() != Some(key.as_slice()) {
            m.due = true;
        }
        m.last = Some(key);
        probing
    })
}

struct MonitorCallback {
    thunk: PostponedThunk,
    handle: SimHandle,
    generation: u64,
}

// SAFETY: see `StrobeCallback`.
unsafe impl Send for MonitorCallback {}

impl super::RegionCallback for MonitorCallback {
    fn invoke(&mut self, ctx: &dyn super::SchedulerCtx) {
        if MONITOR_GENERATION.with(|g| g.get()) != self.generation {
            return; // superseded by a later `$monitor`: retire (no re-arm)
        }
        // Print only if a watched argument signal changed this timestep
        // (a change made while `$monitoroff` is held over until re-enabled).
        // Under the expression-value change test the probes decided whether
        // a display is due; otherwise any watched-signal change is.
        if MONITOR_ENABLED.with(|e| e.get()) {
            let changed = ctx.take_monitor_changed();
            let dedupe = monitor_dedupe_active();
            if (dedupe && monitor_dedupe_take_due()) || (!dedupe && changed) {
                monitor_dedupe_arm(true);
                (self.thunk)(self.handle);
                monitor_dedupe_arm(false);
            }
        }
        ctx.rearm(
            vj_abi::Region::Postponed,
            Box::new(MonitorCallback {
                thunk: self.thunk,
                handle: self.handle,
                generation: self.generation,
            }),
        );
    }
}

/// `$monitor(...)`: make `thunk` the one active monitor. It is checked in the
/// Postponed region of this and every later timestep and prints whenever
/// one of its argument signals -- `slots`, this instance's template slot
/// numbers, `n` of them -- changed value during the timestep (LRM 21.2.3;
/// `$time`-like arguments are not signals, so their change alone never
/// triggers a print).
///
/// # Safety
/// `slots` must point at `n` readable `u32`s (codegen passes a constant
/// global array) and `handle` must be the calling kernel's instance handle.
#[cfg_attr(feature = "aot-exports", unsafe(export_name = "sv_rt_monitor_start"))]
pub unsafe extern "C" fn real_sv_rt_monitor(
    handle: SimHandle,
    thunk: PostponedThunk,
    slots: *const u32,
    n: u32,
) {
    let generation = MONITOR_GENERATION.with(|g| {
        let next = g.get() + 1;
        g.set(next);
        next
    });
    // High bit of `n`: one trailing entry holds the `$time`-like mask and
    // the monitor prints only when an argument expression's value changed.
    let dedupe = n & MONITOR_DEDUPE_N != 0;
    let n = n & !MONITOR_DEDUPE_N;
    let all: &[u32] = if n == 0 && !dedupe {
        &[]
    } else {
        // SAFETY: forwarded to this function's own `# Safety` section.
        unsafe { std::slice::from_raw_parts(slots, n as usize + usize::from(dedupe)) }
    };
    let (slots, mask) = if dedupe {
        (&all[..n as usize], Some(all[n as usize]))
    } else {
        (all, None)
    };
    monitor_dedupe_reset(mask);
    if mask.is_some() {
        MONITOR_DEDUPE.with(|d| {
            if let Some(m) = d.borrow_mut().as_mut() {
                m.probe = Some((thunk, handle));
            }
        });
    }
    let cells: Vec<StateCellId> = slots
        .iter()
        .map(|&s| crate::state::cell_for_slot(handle, vj_abi::SlotId(s)))
        .collect();
    with_current_ops(|ops| {
        ops.set_monitor_watch(cells);
        ops.register_region_callback(
            vj_abi::Region::Postponed,
            ops.now(),
            Box::new(MonitorCallback {
                thunk,
                handle,
                generation,
            }),
        );
    });
}

/// `$monitoron` / `$monitoroff` (LRM 21.2.3).
#[cfg_attr(feature = "aot-exports", unsafe(export_name = "sv_rt_monitor_enable"))]
pub extern "C" fn real_sv_rt_monitor_enable(_handle: SimHandle, on: i32) {
    MONITOR_ENABLED.with(|e| e.set(on != 0));
    // LRM 21.2.3: `$monitoron` produces a display whether or not a value
    // changed (under the expression-value change test; the per-signal form
    // prints only held-over changes).
    if on != 0 {
        MONITOR_DEDUPE.with(|d| {
            if let Some(m) = d.borrow_mut().as_mut() {
                m.due = true;
            }
        });
    }
}

thread_local! {
    /// The time step a concurrent assertion's leaf is being evaluated for, while the
    /// engine re-runs an attempt over its window (`sva::engine`): `$time`/`$realtime`
    /// in an assertion expression is the time of the position the leaf is sampled at
    /// (LRM 1800-2023 16.5.1), not that of the later step doing the re-run.
    static SVA_LEAF_TIME: std::cell::Cell<Option<u64>> = const { std::cell::Cell::new(None) };
}

/// Runs `f` with `$time` answering `ticks` ([`SVA_LEAF_TIME`]).
pub fn with_sva_leaf_time<R>(ticks: u64, f: impl FnOnce() -> R) -> R {
    let prev = SVA_LEAF_TIME.replace(Some(ticks));
    let r = f();
    SVA_LEAF_TIME.set(prev);
    r
}

/// Task F2 (`docs/superpowers/plans/2026-09-27-wave-A2-slice-finish.md`,
/// RT-043 partial): the real production body for the frozen `sv_rt_time`
/// ABI symbol (`vj_codegen::expr::SV_RT_TIME_SYMBOL`) that `$time`/
/// `$stime`/`$realtime` lower to -- reaches the scheduler's real
/// [`SchedulerOps::now`] through the same [`CURRENT_OPS`] thread-local
/// bridge [`real_nba_enqueue`]/[`real_sv_rt_finish`] above already use.
/// Bound the same way as those two (`Jit::add_global_mapping(time_fv,
/// real_sv_rt_time as *const () as usize)`, `vj::run`).
///
/// Takes no `SimHandle` parameter at all -- a deviation both from those two
/// functions' own shape and from the plan's literal `sv_rt_time(state:
/// SimHandle) -> u64` sketch, and load-bearing, not cosmetic:
/// `vj-codegen::expr::fetch_or_declare_sv_rt_time` declares the LLVM
/// side of this ABI boundary as a genuinely zero-argument function (that doc
/// comment records why codegen's call site has no `SimHandle`-carrying value
/// in scope to pass), so this Rust function's own signature must match it
/// exactly -- an extra Rust-side parameter here would desync the `extern
/// "C"` calling convention from the zero-argument call site `build_call`
/// emits (nothing would ever populate it, unlike [`real_sv_rt_finish`],
/// whose `_handle` parameter IS genuinely passed a real argument by its own
/// two-argument call site, just never read). The current time is
/// scheduler-global, not per-instance, so no `SimHandle` is needed here
/// either way.
///
/// A miss (`with_current_ops` finds no `CURRENT_OPS` set) falls through to
/// `0` -- unreachable for every kernel this tree's own codegen emits, per
/// [`with_current_ops`]'s own doc comment, not a silently-wrong nonzero
/// default masking a real bug.
#[cfg_attr(feature = "aot-exports", unsafe(export_name = "sv_rt_time"))]
pub extern "C" fn real_sv_rt_time() -> u64 {
    if let Some(t) = SVA_LEAF_TIME.get() {
        return t;
    }
    let mut ticks: u64 = 0;
    with_current_ops(|ops| ticks = ops.now().ticks);
    ticks
}

/// The reconciliation note this module's own doc comment describes --
/// `sv_kernel_abi_reconciliation_note_exists_and_is_referenced` (below) is
/// the cheap compile-level proof it is not silently dropped in a later
/// refactor.
pub const SV_KERNEL_ABI_RECONCILIATION: &str = "CG-084/CG-111..CG-116 (docs/design/codegen.md §9's SvKernelTable/SvProcessDesc/SvWakeSpec/\
     SvResumeStatus literal draft) is superseded by ARCHITECTURE.md §5's frozen ABI \
     (SimHandle/ProcessFn/EvalResult/Checkpoint), the same reconciliation PR-037/PR-107 already \
     record for the process-model side. CompiledKernelProcess (vj-sim::sched::kernel_adapter) \
     and lower_kernel_fn (vj-codegen::process_lower) implement the frozen shape; the old \
     struct names are never constructed anywhere in this tree.";

#[cfg(test)]
mod tests {
    use super::*;
    use crate::sched::Region;

    /// Minimal `SchedulerOps` double: `CompiledKernelProcess::resume` never
    /// CALLS A METHOD on `ops` itself (it only calls the raw kernel fn
    /// pointer, and, since Task S7, stores `ops`'s address in
    /// `CURRENT_OPS` for that call's duration -- see `with_scheduler_ops`),
    /// so every method here is unreachable in these tests -- kept total (no
    /// `unimplemented!`) since a future `resume` revision that starts using
    /// `ops` should get a real (if trivial) answer, not a panic.
    struct NoOpSchedulerOps;
    impl SchedulerOps for NoOpSchedulerOps {
        fn now(&self) -> crate::sched::SchedTime {
            crate::sched::SchedTime { ticks: 0 }
        }
        fn current_region(&self) -> Region {
            Region::Active
        }
        fn stage_nba(
            &self,
            _cell: StateCellId,
            _value: crate::state::FourStateValue,
            _source: u32,
        ) {
        }
        fn commit_blocking_write(&self, _cell: StateCellId, _value: crate::state::FourStateValue) {}
        fn request_finish(&self, _code: i32) {}
        fn defer_io(
            &self,
            _at: crate::sched::SchedTime,
            _region: Region,
            _cb: crate::sched::DeferredIoCallback,
        ) {
        }
        fn register_region_callback(
            &self,
            _region: Region,
            _at: crate::sched::SchedTime,
            _cb: Box<dyn crate::sched::RegionCallback>,
        ) {
        }
    }

    /// PR-012: `resume_pc == 0` means "fresh/never-run" -- a hand-written
    /// stub kernel asserts it was actually called with `0` on the first call,
    /// and reports a delay suspend the adapter must decode+persist correctly.
    extern "C" fn stub_kernel_initial_dispatch(
        _h: SimHandle,
        _local: *mut u8,
        resume_pc: u32,
    ) -> EvalResult {
        assert_eq!(
            resume_pc, 0,
            "PR-012: the first call to a freshly constructed CompiledKernelProcess must pass \
             resume_pc == 0"
        );
        build_eval_result(ResolvedSuspend::Delay(5), 1)
    }

    #[test]
    fn resume_at_pc_zero_is_initial_dispatch() {
        let mut proc = CompiledKernelProcess::new(
            stub_kernel_initial_dispatch,
            0,
            SimHandle(std::ptr::null_mut()),
        );
        assert_eq!(proc.resume_pc, 0);
        let ops = NoOpSchedulerOps;
        let suspend = proc.resume(SimHandle(std::ptr::null_mut()), ResumeArg::Start, &ops);
        match suspend {
            Suspend::Delay(t) => assert_eq!(t, 5),
            other => panic!("expected Suspend::Delay(5), got {other:?}"),
        }
        assert_eq!(
            proc.resume_pc, 1,
            "the adapter must persist the decoded next resume_pc for the following call"
        );
    }

    #[test]
    fn resume_pc_round_trips_through_evalresult_and_reschedulereq() {
        let er_edge = build_eval_result(
            ResolvedSuspend::Edge {
                cell: StateCellId(7),
                edge: EdgeKind::Posedge,
            },
            3,
        );
        match decode_reschedule(er_edge).expect("KIND_EDGE must decode to Ok") {
            Suspend::WaitEvents(set) => {
                assert_eq!(set.terms.len(), 1);
                match &set.terms[0] {
                    EventTerm::Edge { cell, edge, iff } => {
                        assert_eq!(*cell, StateCellId(7));
                        assert_eq!(*edge, EdgeKind::Posedge);
                        assert!(iff.is_none());
                    }
                    other => panic!("expected EventTerm::Edge, got {other:?}"),
                }
            }
            other => panic!("expected Suspend::WaitEvents, got {other:?}"),
        }

        let er_delay = build_eval_result(ResolvedSuspend::Delay(10), 2);
        match decode_reschedule(er_delay).expect("KIND_DELAY must decode to Ok") {
            Suspend::Delay(t) => assert_eq!(t, 10),
            other => panic!("expected Suspend::Delay(10), got {other:?}"),
        }

        let er_done = build_eval_result(ResolvedSuspend::Done, 0);
        match decode_reschedule(er_done).expect("KIND_DONE must decode to Ok") {
            Suspend::Done => {}
            other => panic!("expected Suspend::Done, got {other:?}"),
        }
    }

    /// Classes plan Task 13: `ResolvedSuspend::ByRequest` (the wire form of
    /// `p.await()` / `p.suspend()`) round-trips to `Suspend::ByRequest` on
    /// its own wire kind -- never aliased to `Done`/`Delay`, which would
    /// terminate or re-time the process instead of resuming it.
    #[test]
    fn by_request_round_trips_through_reschedule_req() {
        let er = build_eval_result(ResolvedSuspend::ByRequest, 9);
        match decode_reschedule(er).expect("KIND_BY_REQUEST must decode to Ok") {
            Suspend::ByRequest => {}
            other => panic!("expected Suspend::ByRequest, got {other:?}"),
        }
    }

    /// Task 6 (CG-081/CG-171): `build_eval_result`/`decode_reschedule` round
    /// trip for `ResolvedSuspend::RepeatEdge` -- the wire tag `SuspendPoint::
    /// CycleDelay` resolves into. Asserts the decoded `Suspend::WaitEvents`
    /// carries exactly one `EventTerm::Repeat{count, inner: Edge{cell,
    /// edge}}` term -- the raw, un-multiplied cycle count, never a
    /// pre-resolved tick offset (ARCHITECTURE §6.5's own "converted at
    /// runtime" requirement, proven bit-for-bit at this wire layer).
    #[test]
    fn repeat_edge_round_trips_through_reschedule_req_as_a_live_repeat_over_edge_term() {
        let er = build_eval_result(
            ResolvedSuspend::RepeatEdge {
                count: 3,
                cell: StateCellId(11),
                edge: EdgeKind::Posedge,
            },
            4,
        );
        match decode_reschedule(er).expect("KIND_REPEAT_EDGE must decode to Ok") {
            Suspend::WaitEvents(set) => {
                assert_eq!(set.terms.len(), 1);
                match &set.terms[0] {
                    EventTerm::Repeat { count, inner } => {
                        assert_eq!(*count, 3);
                        match inner.as_ref() {
                            EventTerm::Edge { cell, edge, iff } => {
                                assert_eq!(*cell, StateCellId(11));
                                assert_eq!(*edge, EdgeKind::Posedge);
                                assert!(iff.is_none());
                            }
                            other => panic!("expected EventTerm::Edge, got {other:?}"),
                        }
                    }
                    other => panic!("expected EventTerm::Repeat, got {other:?}"),
                }
            }
            other => panic!("expected Suspend::WaitEvents, got {other:?}"),
        }
    }

    /// Task 6: the repeat count's full `u64` range survives the wire's `a`/
    /// `b` split -- mirrors `KIND_DELAY`'s own near-`u64::MAX` proof.
    #[test]
    fn repeat_edge_count_round_trips_near_u64_max_through_the_wire() {
        let near_max: u64 = u64::MAX - 1;
        let er = build_eval_result(
            ResolvedSuspend::RepeatEdge {
                count: near_max,
                cell: StateCellId(1),
                edge: EdgeKind::Negedge,
            },
            0,
        );
        match decode_reschedule(er).expect("KIND_REPEAT_EDGE must decode to Ok") {
            Suspend::WaitEvents(set) => match &set.terms[0] {
                EventTerm::Repeat { count, .. } => assert_eq!(*count, near_max),
                other => panic!("expected EventTerm::Repeat, got {other:?}"),
            },
            other => panic!("expected Suspend::WaitEvents, got {other:?}"),
        }
    }

    /// **Review fix (wB-sched Task 5 r1) -- the reachability proof behind
    /// `docs/CONTRACTS.md`'s "Partial closures (wave B)" / SC-068 entry.**
    /// [`ResolvedSuspend::Edge`] (this module's own boundary type, and the
    /// *only* shape `vj-codegen::process_lower::resolve_suspend_point`
    /// -- the one real, non-test producer of a suspend request for a
    /// compiled kernel -- can construct for an edge wait) has no `iff` field
    /// at all, so no real compiled kernel can ever cross this ABI boundary
    /// with one: [`build_eval_result`]/[`decode_reschedule`]'s round trip
    /// always yields `EventTerm::Edge { iff: None, .. }`, regardless of
    /// whether the original SystemVerilog wait was `iff`-qualified
    /// (`@(posedge clk iff en)`). This is why `EngineNotifySink::
    /// notify_change`'s `Some(pred)` filter branch (`sched/control.rs`,
    /// SC-068) is exercised, today, only by a test that hand-builds an
    /// `EventTerm::Edge { iff: Some(_), .. }` directly (e.g. `sched::
    /// control::tests::engine_never_calls_resume_when_iff_is_false_but_does_
    /// on_the_next_true_edge`) -- never by anything this crate's own ABI can
    /// produce from a real kernel. `iff` correctness for a real compiled
    /// kernel instead comes entirely from the separate resume-prologue
    /// mechanism in `vj-codegen::process_lower` (`compile_iff_guard`,
    /// settled by `docs/design/process.md`'s "iff re-check" section) -- the
    /// two mechanisms are not reconciled. If a future change threads `iff`
    /// across this ABI boundary (`RescheduleReqLayoutProposal` gaining a
    /// field, `resolve_suspend_point` populating it), this assertion flips
    /// to `Some` -- that is the documented flip condition; update this test
    /// and the CONTRACTS.md/ledger entries together rather than deleting it.
    ///
    /// **Task 7 (AB-096) status -- this test is still exactly correct,
    /// unchanged:** the ABI-boundary wire representation half of the flip
    /// condition now exists (`RescheduleReqLayoutProposal::guard_fn`/
    /// `ResolvedSuspend::EdgeWithGuard`, proved by
    /// `guard_fn_round_trips_through_reschedule_req` below) and IS reachable
    /// via `decode_reschedule` -- but only for that NEW variant, never for a
    /// plain `ResolvedSuspend::Edge` (still the ONLY shape
    /// `resolve_suspend_point` itself ever constructs today). See
    /// `docs/CONTRACTS.md`'s SC-068 entry, "R-run Task 7 addendum," for the
    /// full accounting of what still blocks `resolve_suspend_point` from
    /// constructing `EdgeWithGuard` for a real compiled kernel.
    #[test]
    fn sc068_decode_reschedule_can_never_produce_a_some_iff_edge_term() {
        let er = build_eval_result(
            ResolvedSuspend::Edge {
                cell: StateCellId(9),
                edge: EdgeKind::Posedge,
            },
            0,
        );
        match decode_reschedule(er).expect("KIND_EDGE must decode to Ok") {
            Suspend::WaitEvents(set) => {
                assert_eq!(set.terms.len(), 1);
                match &set.terms[0] {
                    EventTerm::Edge { iff, .. } => assert!(
                        iff.is_none(),
                        "SC-068 reachability gap regressed: decode_reschedule must never invent \
                         an iff predicate -- ResolvedSuspend::Edge (the only shape a real \
                         compiled kernel's resolve_suspend_point can produce) has no iff field \
                         to carry one across this ABI boundary today"
                    ),
                    other => panic!("expected EventTerm::Edge, got {other:?}"),
                }
            }
            other => panic!("expected Suspend::WaitEvents, got {other:?}"),
        }
    }

    /// A real, top-level `extern "C" fn` for the guard round-trip test below
    /// -- never a closure (no stable, ABI-crossable function pointer of the
    /// `extern "C" fn(SimHandle) -> bool` shape).
    extern "C" fn stub_guard_fn(_h: SimHandle) -> bool {
        false
    }

    /// Task 7 (AB-096): a real `extern "C" fn` pointer round-trips
    /// byte-identical through [`build_eval_result`]/[`decode_reschedule`] --
    /// the literal "documented flip condition"
    /// `sc068_decode_reschedule_can_never_produce_a_some_iff_edge_term`'s
    /// own doc comment names: this is the future change that flips that
    /// assertion to `Some`, for the `ResolvedSuspend::EdgeWithGuard` shape
    /// specifically (a plain `ResolvedSuspend::Edge` still decodes `iff:
    /// None`, proven by that same still-passing, unmodified test).
    #[test]
    fn guard_fn_round_trips_through_reschedule_req() {
        let er = build_eval_result(
            ResolvedSuspend::EdgeWithGuard {
                cell: StateCellId(11),
                edge: EdgeKind::Negedge,
                guard_fn: stub_guard_fn,
            },
            3,
        );
        match decode_reschedule(er).expect("KIND_EDGE (with guard) must decode to Ok") {
            Suspend::WaitEvents(set) => {
                assert_eq!(set.terms.len(), 1);
                match &set.terms[0] {
                    EventTerm::Edge { cell, edge, iff } => {
                        assert_eq!(*cell, StateCellId(11));
                        assert_eq!(*edge, EdgeKind::Negedge);
                        let PredicateFn(decoded) =
                            iff.expect("an EdgeWithGuard suspend must decode iff: Some(..)");
                        assert_eq!(
                            decoded as *const () as usize, stub_guard_fn as *const () as usize,
                            "the decoded guard function pointer must be byte-identical to the \
                             original"
                        );
                    }
                    other => panic!("expected EventTerm::Edge, got {other:?}"),
                }
            }
            other => panic!("expected Suspend::WaitEvents, got {other:?}"),
        }
    }

    /// Review fix (A2-part2 r2, binding rule 8): `decode_reschedule` must
    /// return a typed [`DecodeRescheduleError`], never panic, for
    /// `KIND_EDGE_LIST`/`KIND_FORK_JOIN` -- constructed here directly via
    /// `RescheduleReqLayoutProposal::from_edge_list`/`from_fork_join` (never
    /// through `build_eval_result`, which -- by design, see
    /// `DecodeRescheduleError`'s own doc comment -- never produces these
    /// kinds) and packed into the wire form the same way
    /// `build_eval_result` packs every other kind, so this test exercises
    /// the real decode path, not a hand-rolled shortcut.
    #[test]
    fn decode_reschedule_reports_typed_errors_for_edge_list_and_fork_join() {
        use crate::sched::JoinKind;

        fn wrap(proposal: RescheduleReqLayoutProposal) -> EvalResult {
            let mut opaque = [0u8; vj_abi::eval::RESCHEDULE_REQ_SIZE];
            opaque[..PROPOSAL_SIZE].copy_from_slice(&proposal.to_bytes());
            EvalResult {
                outcome: EvalOutcome::SuspendedOnEvent,
                reschedule: RescheduleReq { opaque },
            }
        }

        let edge_list = wrap(RescheduleReqLayoutProposal::from_edge_list(ArenaRange {
            base: 3,
            count: 5,
        }));
        match decode_reschedule(edge_list) {
            Err(DecodeRescheduleError::UnresolvedEdgeList { base, count }) => {
                assert_eq!(base, 3);
                assert_eq!(count, 5);
            }
            other => panic!(
                "KIND_EDGE_LIST must decode to Err(UnresolvedEdgeList{{3,5}}), never panic; got \
                 {other:?}"
            ),
        }

        let fork_join = wrap(RescheduleReqLayoutProposal::from_fork_join(
            JoinKind::All,
            ArenaRange { base: 10, count: 2 },
        ));
        match decode_reschedule(fork_join) {
            Err(DecodeRescheduleError::UnresolvedForkJoin {
                join_kind,
                base,
                count,
            }) => {
                assert_eq!(join_kind, JoinKind::All as u8);
                assert_eq!(base, 10);
                assert_eq!(count, 2);
            }
            other => panic!(
                "KIND_FORK_JOIN must decode to Err(UnresolvedForkJoin{{..}}), never panic; got \
                 {other:?}"
            ),
        }
    }

    /// The sibling case for a byte outside `KIND_DELAY..=KIND_DONE` entirely
    /// (an ABI/version mismatch) -- also a typed `Err`, never a panic.
    #[test]
    fn decode_reschedule_reports_a_typed_error_for_an_out_of_range_kind_byte() {
        let mut opaque = [0u8; vj_abi::eval::RESCHEDULE_REQ_SIZE];
        opaque[0] = 200;
        let er = EvalResult {
            outcome: EvalOutcome::Completed,
            reschedule: RescheduleReq { opaque },
        };
        match decode_reschedule(er) {
            Err(DecodeRescheduleError::InvalidKind(200)) => {}
            other => panic!("expected Err(InvalidKind(200)), never a panic; got {other:?}"),
        }
    }

    #[test]
    fn sv_kernel_abi_reconciliation_note_exists_and_is_referenced() {
        assert!(
            SV_KERNEL_ABI_RECONCILIATION.contains("CG-084"),
            "the reconciliation note must cite CG-084 so a reader auditing that ledger row \
             finds this exact note"
        );
    }

    /// PR-101/PR-102's structural proof: exactly one `impl CompiledProcess
    /// for` anywhere in this crate's PRODUCTION code (never inside a
    /// `#[cfg(test)]` body, which could trivially inflate the count with a
    /// throwaway test double) bridges a *kernel-backed* type. Mirrors
    /// `sched::process_table`'s own `scan_is_genuinely_slotmap_backed`
    /// pattern: strip every `#[cfg(test)]` block first via the shared
    /// `determinism::scan_support` helper, then search what remains.
    ///
    /// **Documented carve-out:** `sched::control::UnmintedForkChild`
    /// (pre-existing, committed before this task) is also a production
    /// `impl CompiledProcess for` -- but it is a fork/join placeholder
    /// ("a fork child whose real compiled body codegen has not minted yet",
    /// out of the A2 slice's scope entirely, PR-086) that resumes to
    /// `Suspend::Done` unconditionally and bridges no kernel/ABI boundary at
    /// all. It is excluded by name here rather than silently miscounted as a
    /// second kernel adapter -- the two are not the same claim, and pretending
    /// they were would either make this test permanently fail for a reason
    /// unrelated to PR-101/PR-102, or force deleting an unrelated, correct
    /// placeholder this task does not own.
    #[test]
    fn exactly_one_compiledprocess_impl_over_a_kernel_backed_type() {
        const UNRELATED_PLACEHOLDER: &str = "impl CompiledProcess for UnmintedForkChild";

        let src_root = std::path::Path::new(concat!(env!("CARGO_MANIFEST_DIR"), "/src"));
        let mut total = 0usize;
        let mut stack = vec![src_root.to_path_buf()];
        while let Some(dir) = stack.pop() {
            for entry in std::fs::read_dir(&dir).expect("read_dir must succeed under src/") {
                let entry = entry.expect("dir entry must be readable");
                let path = entry.path();
                if path.is_dir() {
                    stack.push(path);
                    continue;
                }
                if path.extension().and_then(|e| e.to_str()) != Some("rs") {
                    continue;
                }
                let text = std::fs::read_to_string(&path)
                    .unwrap_or_else(|e| panic!("must be able to read {path:?}: {e}"));
                let production_only =
                    crate::determinism::scan_support::strip_cfg_test_blocks(&text);
                let here = production_only.matches("impl CompiledProcess for").count()
                    - production_only.matches(UNRELATED_PLACEHOLDER).count();
                total += here;
            }
        }
        assert_eq!(
            total, 1,
            "expected exactly one kernel-backed production `impl CompiledProcess for` in the \
             whole crate (this module's own, on CompiledKernelProcess) -- found {total}"
        );
    }

    /// Task S7: a `SchedulerOps` double that records every `request_finish`
    /// call it receives.
    struct RecordingSchedulerOps {
        finish_calls: std::cell::RefCell<Vec<i32>>,
    }
    impl SchedulerOps for RecordingSchedulerOps {
        fn now(&self) -> crate::sched::SchedTime {
            crate::sched::SchedTime { ticks: 0 }
        }
        fn current_region(&self) -> Region {
            Region::Active
        }
        fn stage_nba(
            &self,
            _cell: StateCellId,
            _value: crate::state::FourStateValue,
            _source: u32,
        ) {
        }
        fn commit_blocking_write(&self, _cell: StateCellId, _value: crate::state::FourStateValue) {}
        fn request_finish(&self, code: i32) {
            self.finish_calls.borrow_mut().push(code);
        }
        fn defer_io(
            &self,
            _at: crate::sched::SchedTime,
            _region: Region,
            _cb: crate::sched::DeferredIoCallback,
        ) {
        }
        fn register_region_callback(
            &self,
            _region: Region,
            _at: crate::sched::SchedTime,
            _cb: Box<dyn crate::sched::RegionCallback>,
        ) {
        }
    }

    /// Task S7 (the `SimHandle`-reachability bridge): `with_scheduler_ops`
    /// makes `ops` reachable to `real_sv_rt_finish` for exactly the duration
    /// of the wrapped call -- proving the actual mechanism `CompiledKernelProcess::
    /// resume` now uses to make `nba_enqueue`/`sv_rt_finish`'s real bodies
    /// reach the scheduler, not merely that `request_finish` itself works.
    #[test]
    fn with_scheduler_ops_makes_ops_reachable_to_real_sv_rt_finish() {
        let ops = RecordingSchedulerOps {
            finish_calls: std::cell::RefCell::new(Vec::new()),
        };
        let handle = SimHandle(std::ptr::null_mut());
        with_scheduler_ops(&ops, || {
            real_sv_rt_finish(handle, 7);
        });
        assert_eq!(
            *ops.finish_calls.borrow(),
            vec![7],
            "real_sv_rt_finish must reach the real ops.request_finish while inside \
             with_scheduler_ops's callback"
        );
    }

    /// Outside any `with_scheduler_ops` call, `real_sv_rt_finish`/
    /// `with_current_ops` must be a silent no-op, never a panic -- proven by
    /// simply calling it with nothing installed and observing no crash (see
    /// `with_current_ops`'s own doc comment for why this shape never arises
    /// from a real kernel).
    #[test]
    fn real_sv_rt_finish_outside_with_scheduler_ops_does_not_panic() {
        let handle = SimHandle(std::ptr::null_mut());
        real_sv_rt_finish(handle, 1);
    }

    /// Review fix (A2-part2 r1, binding rule 7): `with_current_process`
    /// makes `proc` reachable through `CURRENT_PROCESS` for exactly the
    /// duration of the wrapped call, and restores the prior (here: `None`)
    /// value afterward -- the same "installed for one call, then restored"
    /// contract `with_scheduler_ops`/`CURRENT_OPS` already give `ops`,
    /// proven the identical way (`with_scheduler_ops_makes_ops_reachable_to_
    /// real_sv_rt_finish` above).
    #[test]
    fn with_current_process_makes_proc_reachable_for_exactly_one_call_and_restores_after() {
        assert_eq!(
            CURRENT_PROCESS.with(|c| *c.borrow()),
            None,
            "precondition: no other test on this thread left CURRENT_PROCESS set"
        );
        let seen_inside = with_current_process(42, || CURRENT_PROCESS.with(|c| *c.borrow()));
        assert_eq!(
            seen_inside,
            Some(42),
            "CURRENT_PROCESS must read back the ProcessId with_current_process installed, \
             from inside the wrapped call"
        );
        assert_eq!(
            CURRENT_PROCESS.with(|c| *c.borrow()),
            None,
            "with_current_process must restore the prior (None) value after the wrapped call \
             returns, not leave 42 installed"
        );
    }

    /// The save/restore half of the same contract, under nesting: returning
    /// from an inner `with_current_process` must restore the OUTER caller's
    /// `ProcessId`, never merely clear it to `None` -- the exact property
    /// `CURRENT_OPS`'s own doc comment cites as the reason it saves-and-
    /// restores instead of clearing (a hypothetical nested compiled-kernel
    /// call, e.g. a future fork/join child resumed from inside a parent's
    /// own `resume()`, must see ITS OWN `ProcessId` while running, and the
    /// parent's `ProcessId` must still be correct once control returns to
    /// it).
    #[test]
    fn with_current_process_restores_the_outer_value_not_merely_clears_it() {
        with_current_process(1, || {
            let inner = with_current_process(2, || CURRENT_PROCESS.with(|c| *c.borrow()));
            assert_eq!(inner, Some(2));
            assert_eq!(
                CURRENT_PROCESS.with(|c| *c.borrow()),
                Some(1),
                "returning from a nested with_current_process must restore the OUTER caller's \
                 ProcessId, not clear it to None"
            );
        });
        assert_eq!(
            CURRENT_PROCESS.with(|c| *c.borrow()),
            None,
            "after the outermost with_current_process call returns, CURRENT_PROCESS must be \
             back to None"
        );
    }

    /// Review fix (A2-part2 r2, finding 3): `two_process_nba_same_var_
    /// matches_iverilog_bit_for_bit` (`vj-harness`) proves `commit_nba`'s
    /// own push-order hazard resolution (a correct, independent SC-039
    /// design -- see `sched::notify::commit_nba`'s doc comment: it never
    /// reads `NbaUpdate.source`/`entry.source` at all, tie-breaking purely
    /// by push/insertion order). That test's own fixture is a causal
    /// (insertion-order) dependency between its two processes, so it passes
    /// or fails on push-order correctness alone and cannot tell "CURRENT_
    /// PROCESS wired correctly" apart from "CURRENT_PROCESS absent/always-0"
    /// -- reverting the whole `CURRENT_PROCESS` bridge would not change that
    /// test's outcome. This test exercises the one thing r1's fix actually
    /// changed instead: `real_nba_enqueue`'s own `source` output, directly,
    /// with and without `with_current_process` installed -- proving the
    /// bridge is genuinely live, not merely present-but-unread.
    #[test]
    fn real_nba_enqueue_prefers_current_process_over_the_raw_abi_source() {
        // Minimal, frontend-free single-scalar-slot template/instance --
        // `state::instance`'s own doc comments explain why `TemplateRegistry`/
        // `StateLayout`/`SlotDescriptor` need no `sv_lang` dependency at all,
        // so this test builds one without a real `sv_lang::Design`.
        let template = crate::netlist::TemplateKey {
            definition: 1,
            params_hash: 0,
        };
        let layout = crate::state::StateLayout {
            slots: vec![crate::state::SlotDescriptor {
                key: crate::netlist::SymbolKey(0),
                offset: crate::state::SlotOffset {
                    class: crate::state::SlotClass::Inline,
                    word_offset: 0,
                    bit_width: 1,
                    elem_word_count: 1,
                    row_stride_words: 0,
                    num_rows: 0,
                },
                initial: None,
                net_info: None,
            }],
            inline_words: 1,
            wide_words: 0,
        };
        let registry = crate::state::TemplateRegistry::new();
        // SAFETY: single-threaded test; no other live `&StateLayout` borrow
        // and no concurrent `register` call are in flight -- exactly
        // `TemplateRegistry::register`'s own documented obligation.
        unsafe {
            registry.register(template, layout);
        }
        let mut engine = crate::state::StateEngine::new(registry);
        let handle = engine
            .allocate_instance(template)
            .expect("template was just registered above");
        let cell = handle.cell_for_slot(0);
        let h = handle.into_sim_handle();

        struct NbaRecordingOps {
            calls: std::cell::RefCell<Vec<(StateCellId, u32)>>,
        }
        impl SchedulerOps for NbaRecordingOps {
            fn now(&self) -> crate::sched::SchedTime {
                crate::sched::SchedTime { ticks: 0 }
            }
            fn current_region(&self) -> Region {
                Region::Nba
            }
            fn stage_nba(
                &self,
                cell: StateCellId,
                _value: crate::state::FourStateValue,
                source: u32,
            ) {
                self.calls.borrow_mut().push((cell, source));
            }
            fn commit_blocking_write(
                &self,
                _cell: StateCellId,
                _value: crate::state::FourStateValue,
            ) {
            }
            fn request_finish(&self, _code: i32) {}
            fn defer_io(
                &self,
                _at: crate::sched::SchedTime,
                _region: Region,
                _cb: crate::sched::DeferredIoCallback,
            ) {
            }
            fn register_region_callback(
                &self,
                _region: Region,
                _at: crate::sched::SchedTime,
                _cb: Box<dyn crate::sched::RegionCallback>,
            ) {
            }
        }

        let ops = NbaRecordingOps {
            calls: std::cell::RefCell::new(Vec::new()),
        };

        // No CURRENT_PROCESS installed (the shape every direct-call test in
        // this file, bypassing ResumeDriver, produces): real_nba_enqueue
        // must fall back to the raw ABI `source` value (99) it was actually
        // called with.
        with_scheduler_ops(&ops, || {
            real_nba_enqueue(h, vj_abi::SlotId(0), 1, 99);
        });
        assert_eq!(
            *ops.calls.borrow(),
            vec![(cell, 99)],
            "with no CURRENT_PROCESS installed, real_nba_enqueue must fall back to the raw ABI \
             source value"
        );
        ops.calls.borrow_mut().clear();

        // CURRENT_PROCESS installed (exactly as ResumeDriver::resume_for/
        // start install it around every real resume() call): real_nba_enqueue
        // must prefer it (7) over the raw ABI source (99) -- the one
        // observable difference this review fix actually produces.
        with_current_process(7, || {
            with_scheduler_ops(&ops, || {
                real_nba_enqueue(h, vj_abi::SlotId(0), 1, 99);
            });
        });
        assert_eq!(
            *ops.calls.borrow(),
            vec![(cell, 7)],
            "with CURRENT_PROCESS installed, real_nba_enqueue must prefer it (7) over the raw \
             ABI source (99)"
        );

        // SAFETY: reclaims the leaked `Box<InstanceHandle>` `into_sim_handle`
        // minted above -- the same reclaim pattern `state::accessor_impl`'s
        // own tests use after every `into_sim_handle` call.
        unsafe {
            drop(Box::from_raw(h.0 as *mut InstanceHandle));
        }
    }
}