vj-sim 0.3.0

//! SCHED::MATURATION — the generalized "capture now, mature later, flush
//! on resume" deferred-report queue (sva plan T2, §1.4 of
//! `docs/superpowers/plans/2026-10-03-sva.md`).
//!
//! `sched::deferred` originally built this mechanism for exactly one report
//! kind -- deferred immediate assertions (LRM 1800-2023 §16.4.1): a report's
//! text is captured when the assertion's action executes, held keyed by
//! (process, site) with a monotonically increasing generation number, and
//! either written when its maturation region/time arrives (unless a newer
//! report at the same key superseded it first) or discarded entirely if the
//! owning process reaches a flush point first (16.4.2's "a process
//! re-triggered by a glitch discards its pending reports"). This module
//! generalizes that exact mechanism -- unchanged in its own observable
//! behavior for `DeferredImmediate`, proved by `sched::deferred`'s own
//! callers staying byte-identical -- so every later report kind this plan
//! names reuses it instead of a third divergent copy:
//!
//! - `ProceduralConcurrentAssertion` (16.14.1-16.14.4's pass/fail/vacuous
//!   action blocks for an `assert`/`assume`/`cover`/`restrict property`
//!   inside procedural code, T29/T30),
//! - `PropertyPassFail` (a property's own pass/fail action block, T19),
//! - `SequenceCoverMatch` (16.14.3's `cover sequence` match bookkeeping,
//!   T32),
//! - `CheckerAction` (17.7.3's checker action-block scheduling semantics,
//!   T39).
//!
//! **Deviation from the plan's literal `MaturationQueue<DeferredImmediate>`
//! generic-type-parameter phrasing:** this implementation parameterizes by
//! an enum *value* (`ReportKind`), not a type parameter instantiated as
//! `MaturationQueue::<DeferredImmediate>`. A type-parameter design was
//! prototyped and rejected: its natural shape is `Mature<K>` holding a
//! `&'static MaturationQueue<K>` back-reference so `invoke` can reach the
//! right storage, but `MaturationQueue<K>`'s `RefCell`/`Cell` fields are not
//! `Sync`, so `&'static MaturationQueue<K>` is not `Send` -- and
//! `RegionCallback: Send` is a hard requirement (`region_callback.rs`).
//! Keeping ONE shared `thread_local!` map, keyed by `(ReportKind, Option
//! <ProcessId>, u64)`, avoids that soundness hole entirely (a `Mature`
//! value is then just three `Copy` fields, trivially `Send`) while still
//! giving every later report kind its own disjoint key space -- the literal
//! LRM behavior this task's own acceptance text asks for ("same PENDING/
//! GENERATION/flush-point API, parameterized") is preserved exactly; only
//! the Rust-level mechanism of parameterization differs.

use std::cell::{Cell, RefCell};
use std::collections::HashMap;

use vj_abi::Region;

use crate::sched::{ProcessId, RegionCallback, SchedulerCtx, SimTick};

/// Which subsystem produced a report maturing through this module -- part
/// of every report's key (see this module's own doc comment for why), so
/// two different kinds reporting at the identical (process, site) pair
/// (coincidence, not a collision -- an assertion and an unrelated checker
/// action could legitimately share a source offset across different files'
/// line-counting, however unlikely) never mature, supersede, or flush each
/// other.
#[derive(Clone, Copy, PartialEq, Eq, Debug, Hash)]
pub enum ReportKind {
    /// LRM §16.4 (`assert #0`/`assert final`) -- `sched::deferred`'s own,
    /// only wired-up kind today.
    DeferredImmediate,
    /// LRM §16.14.1-16.14.4 -- a procedural concurrent assertion's own
    /// pass/fail/vacuous action block (T29/T30; not wired up yet).
    ProceduralConcurrentAssertion,
    /// LRM §16.14.1 -- a property's own pass/fail action block (T19; not
    /// wired up yet).
    PropertyPassFail,
    /// LRM §16.14.3 -- a `cover sequence` match (T32; not wired up yet).
    SequenceCoverMatch,
    /// LRM §17.7.3 -- a checker action-block (T39; not wired up yet).
    CheckerAction,
    /// LRM §16.9.4 -- the action block (or default `$error`) of an assertion
    /// containing global clocking future sampled value functions, held until
    /// the global clock tick that follows the attempt's last assertion-clock
    /// tick ([`report_at_global_tick`]; T18, the engine T29 wires it up).
    GlobalClockDelayed,
    /// LRM §12.4.2.1/§12.5.3.1 -- a `unique`/`unique0`/`priority` `if` or
    /// `case` violation report: checked when the statement executes, held on
    /// the executing process's violation report queue and matured in the
    /// Observed region of that time step ([`report_independent`]), discarded
    /// instead if the process reaches a violation report flush point first
    /// ([`flush_violations`]).
    Violation,
    /// LRM §16.4.1 -- the action of a deferred assertion that is a call of a
    /// user task or void function: the call (a compiled thunk plus its
    /// arguments, captured by value when the assertion executed) is made when
    /// the report matures, not held as text. Flushed like
    /// [`Self::DeferredImmediate`] (16.4.2); its payload is [`call_payload`]'s.
    DeferredCall,
}

/// Where a matured report's captured bytes go.
pub type Writer = fn(&[u8]);

/// One report's key: which kind produced it, the owning process (`None`
/// outside one), and the site identifying *which* report (an assertion's
/// source offset, today).
type Key = (ReportKind, Option<ProcessId>, u64);

thread_local! {
    /// Every kind's pending reports share one map, keyed by `(kind, proc,
    /// site)` -- a shared map keyed by kind, rather than N separate maps,
    /// is both correct (no cross-kind collision: the key includes `kind`)
    /// and simpler (one `thread_local!`, not a hand-rolled one per kind).
    static PENDING: RefCell<HashMap<Key, (u64, Vec<u8>)>> = RefCell::new(HashMap::new());
    static GENERATION: Cell<u64> = const { Cell::new(0) };
    /// Each kind's own writer. A kind nothing ever calls `set_writer` for
    /// simply never writes its matured reports (the same precondition
    /// `sched::deferred`'s own single-kind `set_deferred_writer` already
    /// placed on its one caller, `vj::run`) -- never a panic, since
    /// a report kind with no runtime wiring yet (every kind but
    /// `DeferredImmediate`, this task) legitimately has no writer.
    static WRITERS: RefCell<HashMap<ReportKind, Writer>> = RefCell::new(HashMap::new());
    /// `GlobalClockDelayed` reports not yet handed to a region callback, in
    /// capture order: `(key, generation, simulation time they were captured
    /// at)`. They mature at a global clock tick, which only the global clock
    /// itself can announce ([`global_tick`]).
    static AWAITING: RefCell<Vec<(Key, u64, SimTick)>> = const { RefCell::new(Vec::new()) };
    /// The assertion control site ([`crate::sched::assert_control`]) a pending
    /// report belongs to, for the reports reported with one: `$assertkill`
    /// discards the queued reports of the sites it reaches ([`flush_sites`]).
    /// Entries follow their report: removed when it matures or is discarded.
    static SITES: RefCell<HashMap<Key, u32>> = RefCell::new(HashMap::new());
}

/// Sets where `kind`'s matured reports are written.
pub fn set_writer(kind: ReportKind, f: Writer) {
    WRITERS.with(|w| w.borrow_mut().insert(kind, f));
}

/// Holds `text` -- `kind`'s just-captured output at source offset `site`,
/// produced by the process currently running (none outside one) -- until
/// `region` of the current time step, when it matures (see [`Mature::
/// invoke`]) unless flushed ([`flush`]) or superseded by a newer report at
/// the same key first.
pub fn report(kind: ReportKind, site: u64, region: Region, text: Vec<u8>) {
    report_for_site(kind, site, region, text, None);
}

/// [`report`] for the assertion whose control site is `control_site`: a
/// `$assertkill` that reaches the site discards the report if it is still
/// pending ([`flush_sites`]).
pub fn report_for_site(
    kind: ReportKind,
    site: u64,
    region: Region,
    text: Vec<u8>,
    control_site: Option<u32>,
) {
    let key = (kind, crate::sched::current_process(), site);
    let generation = GENERATION.with(|g| {
        g.set(g.get() + 1);
        g.get()
    });
    PENDING.with(|p| p.borrow_mut().insert(key, (generation, text)));
    tag_site(key, control_site);
    crate::sched::with_current_ops(|ops| {
        ops.register_region_callback(region, ops.now(), Box::new(Mature { key, generation }));
    });
}

/// Holds `text` -- the report of an assertion attempt that ended in this time
/// step but contains global clocking future sampled value functions (LRM
/// 16.9.4) -- until the first global clock tick STRICTLY after now, when
/// [`global_tick`] matures it. The text is captured now, so a message that
/// names the attempt's end time names the attempt's time (80 in the LRM's `a3`
/// example), not the time the report is finally written (90).
///
/// Unlike [`report`], every call is its own report: each failing attempt runs
/// its own action block, so a later attempt must not supersede an earlier one
/// still waiting for the tick. Not discarded by [`flush`] either: 16.9.4 says
/// `disable iff` and the other asynchronous controls act only within the
/// attempt's interval, and an attempt that has ended is no longer affected.
pub fn report_at_global_tick(text: Vec<u8>) {
    crate::sched::with_current_ops(|ops| {
        hold_for_global_tick(text, ops.now().ticks);
    });
}

/// [`report_at_global_tick`] with the capture time explicit.
fn hold_for_global_tick(text: Vec<u8>, captured: SimTick) {
    let generation = GENERATION.with(|g| {
        g.set(g.get() + 1);
        g.get()
    });
    // The generation is the site: no two holds share a key (see above).
    let key = (
        ReportKind::GlobalClockDelayed,
        crate::sched::current_process(),
        generation,
    );
    PENDING.with(|p| p.borrow_mut().insert(key, (generation, text)));
    AWAITING.with(|a| a.borrow_mut().push((key, generation, captured)));
}

/// Holds `text` -- a violation report ([`ReportKind::Violation`]) the running
/// process just detected -- until `region` of the current time step, unless
/// the process reaches a violation report flush point first.
///
/// Unlike [`report`], every call is its own report: a statement that executes
/// twice in one time step (a loop body, a re-run process) has each execution
/// checked and reported independently (LRM 12.4.2.1: "any iteration in the loop
/// can report a uniqueness violation"), so a later report never supersedes an
/// earlier one still pending.
pub fn report_independent(kind: ReportKind, region: Region, text: Vec<u8>) {
    report_independent_for_site(kind, region, text, None);
}

/// [`report_independent`] for the assertion whose control site is
/// `control_site` (see [`report_for_site`]).
pub fn report_independent_for_site(
    kind: ReportKind,
    region: Region,
    text: Vec<u8>,
    control_site: Option<u32>,
) {
    let generation = GENERATION.with(|g| {
        g.set(g.get() + 1);
        g.get()
    });
    // The generation is the site: no two reports share a key.
    let key = (kind, crate::sched::current_process(), generation);
    PENDING.with(|p| p.borrow_mut().insert(key, (generation, text)));
    tag_site(key, control_site);
    crate::sched::with_current_ops(|ops| {
        ops.register_region_callback(region, ops.now(), Box::new(Mature { key, generation }));
    });
}

/// Records which control site the report at `key` belongs to.
fn tag_site(key: Key, control_site: Option<u32>) {
    match control_site {
        Some(site) => SITES.with(|s| {
            s.borrow_mut().insert(key, site);
        }),
        // A newer report at the same key (deferred reports supersede) replaces
        // the old one, so the old one's site goes with it.
        None => SITES.with(|s| {
            s.borrow_mut().remove(&key);
        }),
    }
}

/// Drops the site tags of reports no longer pending.
fn prune_sites() {
    SITES.with(|s| {
        let mut s = s.borrow_mut();
        if !s.is_empty() {
            PENDING.with(|p| {
                let p = p.borrow();
                s.retain(|k, _| p.contains_key(k));
            });
        }
    });
}

/// Discards every pending report (any process, any kind) that was reported for
/// one of the control sites `sites`: `$assertkill` flushes the queued deferred
/// reports and queued violation reports of the assertions it reaches (LRM
/// 20.11).
pub fn flush_sites(sites: &[u32]) {
    let doomed: Vec<Key> = SITES.with(|s| {
        s.borrow()
            .iter()
            .filter(|(_, site)| sites.contains(site))
            .map(|(k, _)| *k)
            .collect()
    });
    PENDING.with(|p| {
        let mut p = p.borrow_mut();
        for k in &doomed {
            p.remove(k);
        }
    });
    prune_sites();
}

/// Discards `proc`'s pending reports of the control site `site`: `disable
/// <deferred assertion label>` cancels "any pending assertion reports for that
/// assertion" of the executing process only (LRM 16.4.4; other processes'
/// reports of the same assertion are independent, 16.4.5).
pub fn flush_process_site(proc: ProcessId, site: u32) {
    let doomed: Vec<Key> = SITES.with(|s| {
        s.borrow()
            .iter()
            .filter(|(k, v)| **v == site && k.1 == Some(proc))
            .map(|(k, _)| *k)
            .collect()
    });
    PENDING.with(|p| {
        let mut p = p.borrow_mut();
        for k in &doomed {
            p.remove(k);
        }
    });
    prune_sites();
}

/// Removes and returns, in capture order, every held report captured strictly
/// before `now`: a report captured in the very time step of a global tick
/// waits for the NEXT one (16.9.4: "the global clocking tick that follows").
fn take_due(now: SimTick) -> Vec<(Key, u64)> {
    AWAITING.with(|a| {
        let mut a = a.borrow_mut();
        let mut due = Vec::new();
        a.retain(|&(key, generation, captured)| {
            if captured < now {
                due.push((key, generation));
                false
            } else {
                true
            }
        });
        due
    })
}

/// The global clock ticked in this time step: matures, in `region`, every
/// report [`report_at_global_tick`] captured in an earlier one.
pub fn global_tick(region: Region) {
    crate::sched::with_current_ops(|ops| {
        for (key, generation) in take_due(ops.now().ticks) {
            ops.register_region_callback(region, ops.now(), Box::new(Mature { key, generation }));
        }
    });
}

/// Discards every one of `proc`'s pending reports, of any kind (it reached
/// a flush point). `sched::deferred`'s original `flush_deferred` only ever
/// needed to flush `DeferredImmediate` reports (the only kind that
/// existed); flushing across every kind here is this generalization's
/// whole point -- 16.4.2's "reaching a flush point discards pending
/// reports" rule applies identically to every later kind's own flush-point
/// rule (16.14.6.3's procedural concurrent assertion glitch rule, 17.7.3's
/// checker action-block rule), not only deferred immediate assertions'.
/// Behavior-preserving for today's one real caller (`kernel_adapter.rs`'s
/// flush-point hook): with only `DeferredImmediate` ever reported, flushing
/// "every kind" and flushing "just that kind" are the identical operation.
pub fn flush(proc: ProcessId) {
    // `GlobalClockDelayed` reports belong to attempts that have already ended
    // (see `report_at_global_tick`), so a flush point does not cancel them.
    // `Violation` reports have a flush rule of their own ([`flush_violations`]):
    // only an event control or wait resumption discards them, never a delay.
    PENDING.with(|p| {
        p.borrow_mut().retain(|k, _| {
            k.1 != Some(proc)
                || matches!(k.0, ReportKind::GlobalClockDelayed | ReportKind::Violation)
        })
    });
    prune_sites();
}

/// Discards `proc`'s pending violation reports ([`ReportKind::Violation`]):
/// it reached a violation report flush point (LRM 12.4.2.1) -- resumed from an
/// event control or `wait`, or an `always_comb`/`always_latch` was triggered
/// again by a transition on one of its dependent signals.
pub fn flush_violations(proc: ProcessId) {
    PENDING.with(|p| {
        p.borrow_mut()
            .retain(|k, _| k.1 != Some(proc) || k.0 != ReportKind::Violation)
    });
    prune_sites();
}

/// How many reports are pending (tests of the report producers).
#[cfg(test)]
pub(crate) fn pending_len() -> usize {
    PENDING.with(|p| p.borrow().len())
}

/// The payload of a [`ReportKind::DeferredCall`] report: the thunk's address,
/// the instance handle it runs in, then the arguments' bytes
/// (`thunk(handle, args)`, `sv_rt_deferred_call`).
pub fn call_payload(thunk: usize, handle: usize, args: &[u8]) -> Vec<u8> {
    let mut p = Vec::with_capacity(16 + args.len());
    p.extend_from_slice(&(thunk as u64).to_le_bytes());
    p.extend_from_slice(&(handle as u64).to_le_bytes());
    p.extend_from_slice(args);
    p
}

/// Makes the call a [`call_payload`] holds (LRM 16.4.1: when the report
/// matures).
fn run_call(payload: &[u8]) {
    type Thunk = unsafe extern "C" fn(usize, *const u128);
    if payload.len() < 16 {
        return;
    }
    let word = |i: usize| u64::from_le_bytes(payload[i..i + 8].try_into().unwrap());
    let (thunk, handle) = (word(0) as usize, word(8) as usize);
    // The arguments are laid out as the thunk's LLVM struct: copy them to
    // memory aligned for any member (an `i128` needs 16).
    let args = &payload[16..];
    let mut buf = vec![0u128; args.len().div_ceil(16).max(1)];
    // SAFETY: `buf` holds at least `args.len()` bytes; the regions are distinct.
    unsafe {
        std::ptr::copy_nonoverlapping(args.as_ptr(), buf.as_mut_ptr().cast::<u8>(), args.len());
    }
    // SAFETY: `thunk` is the address of the compiled `__vj_deferred$N`
    // function the queueing site embedded (`void(ptr handle, ptr args)`), and
    // `handle` the instance it was queued from.
    unsafe {
        let f: Thunk = std::mem::transmute::<usize, Thunk>(thunk);
        f(handle, buf.as_ptr());
    }
}

/// A report maturing: written unless flushed or replaced since.
struct Mature {
    key: Key,
    generation: u64,
}

impl RegionCallback for Mature {
    fn invoke(&mut self, _ctx: &dyn SchedulerCtx) {
        let text = PENDING.with(|p| {
            let mut p = p.borrow_mut();
            match p.get(&self.key) {
                Some((g, _)) if *g == self.generation => p.remove(&self.key).map(|(_, t)| t),
                _ => None,
            }
        });
        if let Some(text) = text {
            SITES.with(|s| {
                s.borrow_mut().remove(&self.key);
            });
            let kind = self.key.0;
            if kind == ReportKind::DeferredCall {
                run_call(&text);
                return;
            }
            let writer = WRITERS.with(|w| w.borrow().get(&kind).copied());
            if let Some(w) = writer {
                w(&text);
            }
        }
    }
}

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

    /// Every existing `sched::deferred` unit test must still pass
    /// unmodified (this task's own acceptance text) -- `sched::deferred` has
    /// no `#[cfg(test)]` module of its own (it is exercised end-to-end via
    /// the `deferred_assertions.sv` probe-corpus fixture, per `sched::
    /// deferred`'s own module doc comment history), so that "unmodified"
    /// guarantee is proved by this module's own direct, generic-mechanism
    /// tests below plus the unchanged probe-corpus fixture, not a moved
    /// file's test functions.
    fn reset_thread_locals() {
        PENDING.with(|p| p.borrow_mut().clear());
        GENERATION.with(|g| g.set(0));
        WRITERS.with(|w| w.borrow_mut().clear());
        AWAITING.with(|a| a.borrow_mut().clear());
        SITES.with(|s| s.borrow_mut().clear());
    }

    thread_local! {
        static CAPTURED: RefCell<Vec<u8>> = const { RefCell::new(Vec::new()) };
    }

    fn capture_writer(bytes: &[u8]) {
        CAPTURED.with(|c| c.borrow_mut().extend_from_slice(bytes));
    }

    /// This task's own named test: two different `ReportKind`s reporting
    /// at the IDENTICAL (process, site) key must never collide -- each
    /// kind's maturation is independent, proving the generalization's
    /// disjoint-key-space property (this module's own doc comment,
    /// "Deviation" paragraph) rather than merely asserting the types exist.
    /// Runs entirely outside the scheduler (no `with_current_ops` context,
    /// no region callback dispatch) by calling `Mature::invoke` directly --
    /// `report`'s own region-callback registration half is covered
    /// end-to-end by the existing `deferred_assertions.sv` probe-corpus
    /// fixture and `sched::control`'s own Observed-region tests instead.
    #[test]
    fn maturation_queue_generic_over_report_kind_still_separates_by_key() {
        reset_thread_locals();
        CAPTURED.with(|c| c.borrow_mut().clear());
        set_writer(ReportKind::DeferredImmediate, capture_writer);
        set_writer(ReportKind::ProceduralConcurrentAssertion, capture_writer);

        // Same (proc=None, site=42) key, two different kinds.
        let key_a = (ReportKind::DeferredImmediate, None, 42u64);
        let key_b = (ReportKind::ProceduralConcurrentAssertion, None, 42u64);
        let gen_a = GENERATION.with(|g| {
            g.set(g.get() + 1);
            g.get()
        });
        PENDING.with(|p| {
            p.borrow_mut()
                .insert(key_a, (gen_a, b"deferred-immediate".to_vec()))
        });
        let gen_b = GENERATION.with(|g| {
            g.set(g.get() + 1);
            g.get()
        });
        PENDING.with(|p| {
            p.borrow_mut()
                .insert(key_b, (gen_b, b"procedural-concurrent-assertion".to_vec()))
        });

        // Maturing kind A must not touch kind B's still-pending entry.
        let mut mature_a = Mature {
            key: key_a,
            generation: gen_a,
        };
        struct NoopCtx;
        impl SchedulerCtx for NoopCtx {
            fn now(&self) -> crate::sched::SchedTime {
                crate::sched::SchedTime { ticks: 0 }
            }
            fn read(&self, _cell: crate::state::StateCellId) -> crate::state::FourStateValue {
                panic!("not used by this test")
            }
            fn rearm(&self, _region: Region, _cb: Box<dyn RegionCallback>) {
                panic!("not used by this test")
            }
            fn take_monitor_changed(&self) -> bool {
                panic!("not used by this test")
            }
        }
        mature_a.invoke(&NoopCtx);

        assert_eq!(
            CAPTURED.with(|c| c.borrow().clone()),
            b"deferred-immediate",
            "kind A's own write must have fired"
        );
        assert!(
            PENDING.with(|p| p.borrow().contains_key(&key_b)),
            "kind B's pending entry at the identical (proc, site) must survive kind A's own \
             maturation untouched -- the key's kind component is what keeps them disjoint"
        );

        // Maturing kind B now must fire its own distinct text.
        CAPTURED.with(|c| c.borrow_mut().clear());
        let mut mature_b = Mature {
            key: key_b,
            generation: gen_b,
        };
        mature_b.invoke(&NoopCtx);
        assert_eq!(
            CAPTURED.with(|c| c.borrow().clone()),
            b"procedural-concurrent-assertion"
        );
        assert!(!PENDING.with(|p| p.borrow().contains_key(&key_b)));

        reset_thread_locals();
    }

    /// `flush` must discard every kind's pending reports for the flushed
    /// process, not just one kind's -- this generalization's own "flush
    /// across all kinds" decision (this module's own doc comment on
    /// [`flush`]).
    #[test]
    fn flush_discards_every_kind_for_the_flushed_process() {
        reset_thread_locals();
        let proc: ProcessId = 7;
        PENDING.with(|p| {
            let mut p = p.borrow_mut();
            p.insert(
                (ReportKind::DeferredImmediate, Some(proc), 1),
                (1, b"a".to_vec()),
            );
            p.insert(
                (ReportKind::CheckerAction, Some(proc), 2),
                (2, b"b".to_vec()),
            );
            p.insert(
                (ReportKind::DeferredImmediate, Some(99), 3),
                (3, b"other-proc".to_vec()),
            );
        });

        flush(proc);

        PENDING.with(|p| {
            let p = p.borrow();
            assert!(!p.contains_key(&(ReportKind::DeferredImmediate, Some(proc), 1)));
            assert!(!p.contains_key(&(ReportKind::CheckerAction, Some(proc), 2)));
            assert!(
                p.contains_key(&(ReportKind::DeferredImmediate, Some(99), 3)),
                "a different process's pending report must survive this flush"
            );
        });

        reset_thread_locals();
    }

    /// A kind with no `set_writer` call ever made for it must simply not
    /// write -- never panic -- when one of its reports matures (this
    /// module's own doc comment on the `WRITERS` thread-local).
    #[test]
    fn maturing_a_kind_with_no_registered_writer_does_not_panic() {
        reset_thread_locals();
        let key = (ReportKind::SequenceCoverMatch, None, 5u64);
        let generation = GENERATION.with(|g| {
            g.set(g.get() + 1);
            g.get()
        });
        PENDING.with(|p| p.borrow_mut().insert(key, (generation, b"x".to_vec())));

        let mut mature = Mature { key, generation };
        struct NoopCtx;
        impl SchedulerCtx for NoopCtx {
            fn now(&self) -> crate::sched::SchedTime {
                crate::sched::SchedTime { ticks: 0 }
            }
            fn read(&self, _cell: crate::state::StateCellId) -> crate::state::FourStateValue {
                panic!("not used by this test")
            }
            fn rearm(&self, _region: Region, _cb: Box<dyn RegionCallback>) {
                panic!("not used by this test")
            }
            fn take_monitor_changed(&self) -> bool {
                panic!("not used by this test")
            }
        }
        mature.invoke(&NoopCtx); // Must not panic.

        assert!(
            !PENDING.with(|p| p.borrow().contains_key(&key)),
            "the entry is still consumed (removed) even with no writer registered"
        );

        reset_thread_locals();
    }

    /// A newer report at the same key supersedes an older one still
    /// pending: maturing the OLDER report's own `Mature` callback (stale
    /// generation) must not fire at all, and must not disturb the newer
    /// one's own pending entry -- the identical "replaces its pending
    /// report" rule `sched::deferred`'s own module doc comment states for
    /// `DeferredImmediate`, proved here generically.
    #[test]
    fn a_stale_generation_mature_callback_does_not_fire() {
        reset_thread_locals();
        // A plain `fn` pointer writer can't close over a local log, so this
        // test distinguishes "fired" from "did not fire" by checking
        // PENDING's own post-state instead, mirroring `maturing_a_kind_
        // with_no_registered_writer_does_not_panic`'s approach.
        let key = (ReportKind::PropertyPassFail, None, 9u64);
        let stale_generation = GENERATION.with(|g| {
            g.set(g.get() + 1);
            g.get()
        });
        PENDING.with(|p| {
            p.borrow_mut()
                .insert(key, (stale_generation, b"old".to_vec()))
        });
        // A newer report at the identical key supersedes it.
        let fresh_generation = GENERATION.with(|g| {
            g.set(g.get() + 1);
            g.get()
        });
        PENDING.with(|p| {
            p.borrow_mut()
                .insert(key, (fresh_generation, b"new".to_vec()))
        });

        let mut stale = Mature {
            key,
            generation: stale_generation,
        };
        struct NoopCtx;
        impl SchedulerCtx for NoopCtx {
            fn now(&self) -> crate::sched::SchedTime {
                crate::sched::SchedTime { ticks: 0 }
            }
            fn read(&self, _cell: crate::state::StateCellId) -> crate::state::FourStateValue {
                panic!("not used by this test")
            }
            fn rearm(&self, _region: Region, _cb: Box<dyn RegionCallback>) {
                panic!("not used by this test")
            }
            fn take_monitor_changed(&self) -> bool {
                panic!("not used by this test")
            }
        }
        stale.invoke(&NoopCtx);

        PENDING.with(|p| {
            let p = p.borrow();
            let (g, text) = p.get(&key).expect("the fresh report must still be pending");
            assert_eq!(*g, fresh_generation);
            assert_eq!(text, b"new");
        });

        reset_thread_locals();
    }

    /// What `global_tick` does at time `now`, minus the scheduler: mature
    /// every report `take_due(now)` returns, directly.
    fn mature_due_at(now: SimTick) {
        struct NoopCtx;
        impl SchedulerCtx for NoopCtx {
            fn now(&self) -> crate::sched::SchedTime {
                crate::sched::SchedTime { ticks: 0 }
            }
            fn read(&self, _cell: crate::state::StateCellId) -> crate::state::FourStateValue {
                panic!("not used by this test")
            }
            fn rearm(&self, _region: Region, _cb: Box<dyn RegionCallback>) {
                panic!("not used by this test")
            }
            fn take_monitor_changed(&self) -> bool {
                panic!("not used by this test")
            }
        }
        for (key, generation) in take_due(now) {
            Mature { key, generation }.invoke(&NoopCtx);
        }
    }

    fn captured() -> String {
        String::from_utf8(CAPTURED.with(|c| c.borrow().clone())).unwrap()
    }

    /// LRM 16.9.4's `a3`: the attempt fails at time 80, a global tick; its
    /// `$error` runs at the NEXT global tick, 90, and still reports 80. The
    /// report is not due in the time step it was captured in -- even though
    /// 80 is itself a global tick -- and not at any tick before 90.
    #[test]
    fn a_global_tick_report_matures_at_the_next_tick_and_keeps_its_own_time() {
        reset_thread_locals();
        CAPTURED.with(|c| c.borrow_mut().clear());
        set_writer(ReportKind::GlobalClockDelayed, capture_writer);

        hold_for_global_tick(b"sig is not stable (attempt ended at 80)\n".to_vec(), 80);
        mature_due_at(80);
        assert_eq!(
            captured(),
            "",
            "the tick of the capture time step is not the next"
        );
        mature_due_at(90);
        assert_eq!(captured(), "sig is not stable (attempt ended at 80)\n");

        reset_thread_locals();
    }

    /// Reports held at different times mature in capture order, each at the
    /// first tick strictly after its own capture, and only once.
    #[test]
    fn global_tick_reports_mature_in_capture_order_once() {
        reset_thread_locals();
        CAPTURED.with(|c| c.borrow_mut().clear());
        set_writer(ReportKind::GlobalClockDelayed, capture_writer);

        hold_for_global_tick(b"A@80\n".to_vec(), 80);
        hold_for_global_tick(b"B@85\n".to_vec(), 85);
        hold_for_global_tick(b"C@90\n".to_vec(), 90);
        mature_due_at(90);
        assert_eq!(
            captured(),
            "A@80\nB@85\n",
            "C was captured in tick 90's own step"
        );
        mature_due_at(90);
        assert_eq!(captured(), "A@80\nB@85\n", "a report matures once");
        mature_due_at(100);
        assert_eq!(captured(), "A@80\nB@85\nC@90\n");

        reset_thread_locals();
    }

    /// Two attempts of one assertion that both fail before the next tick are
    /// two reports: nothing supersedes (the opposite of a deferred report's
    /// rule).
    #[test]
    fn global_tick_reports_never_supersede_each_other() {
        reset_thread_locals();
        CAPTURED.with(|c| c.borrow_mut().clear());
        set_writer(ReportKind::GlobalClockDelayed, capture_writer);

        hold_for_global_tick(b"first\n".to_vec(), 71);
        hold_for_global_tick(b"second\n".to_vec(), 72);
        mature_due_at(80);
        assert_eq!(captured(), "first\nsecond\n");

        reset_thread_locals();
    }

    /// 16.9.4 / `a3`: an attempt that has ended is not cancelled by a later
    /// disable, so a process's flush point leaves its held reports alone,
    /// while still discarding every other kind.
    #[test]
    fn a_flush_point_does_not_cancel_a_held_global_tick_report() {
        reset_thread_locals();
        CAPTURED.with(|c| c.borrow_mut().clear());
        set_writer(ReportKind::GlobalClockDelayed, capture_writer);
        set_writer(ReportKind::PropertyPassFail, capture_writer);

        let proc: ProcessId = 3;
        crate::sched::with_current_process(proc, || {
            hold_for_global_tick(b"held\n".to_vec(), 80);
        });
        PENDING.with(|p| {
            p.borrow_mut().insert(
                (ReportKind::PropertyPassFail, Some(proc), 1),
                (99, b"gone\n".to_vec()),
            )
        });
        flush(proc);

        mature_due_at(90);
        assert_eq!(captured(), "held\n");
        assert!(
            PENDING.with(|p| p.borrow().is_empty()),
            "the other kind was flushed"
        );

        reset_thread_locals();
    }

    /// Matures every pending report, in capture order, as the Observed region
    /// callbacks `report_independent` registers would.
    fn mature_all_pending() {
        struct NoopCtx;
        impl SchedulerCtx for NoopCtx {
            fn now(&self) -> crate::sched::SchedTime {
                crate::sched::SchedTime { ticks: 0 }
            }
            fn read(&self, _cell: crate::state::StateCellId) -> crate::state::FourStateValue {
                panic!("not used by this test")
            }
            fn rearm(&self, _region: Region, _cb: Box<dyn RegionCallback>) {
                panic!("not used by this test")
            }
            fn take_monitor_changed(&self) -> bool {
                panic!("not used by this test")
            }
        }
        let mut due: Vec<(Key, u64)> = PENDING.with(|p| {
            p.borrow()
                .iter()
                .map(|(k, (g, _))| (*k, *g))
                .collect::<Vec<_>>()
        });
        due.sort_by_key(|&(_, g)| g);
        for (key, generation) in due {
            Mature { key, generation }.invoke(&NoopCtx);
        }
    }

    /// LRM 12.4.2.1: every execution of a checked statement reports on its own.
    /// Two reports of one process never supersede each other (unlike `report`,
    /// whose same-site report replaces the earlier one).
    #[test]
    fn independent_reports_of_one_process_never_supersede_each_other() {
        reset_thread_locals();
        CAPTURED.with(|c| c.borrow_mut().clear());
        set_writer(ReportKind::Violation, capture_writer);

        crate::sched::with_current_process(4, || {
            report_independent(ReportKind::Violation, Region::Observed, b"first\n".to_vec());
            report_independent(
                ReportKind::Violation,
                Region::Observed,
                b"second\n".to_vec(),
            );
        });
        mature_all_pending();
        assert_eq!(captured(), "first\nsecond\n");

        reset_thread_locals();
    }

    /// The violation flush point discards only the flushed process's
    /// `Violation` reports: not another process's, and not that process's other
    /// kinds (a deferred immediate assertion has its own flush rule).
    #[test]
    fn flush_violations_discards_only_that_processs_violation_reports() {
        reset_thread_locals();
        CAPTURED.with(|c| c.borrow_mut().clear());
        set_writer(ReportKind::Violation, capture_writer);
        set_writer(ReportKind::DeferredImmediate, capture_writer);

        crate::sched::with_current_process(4, || {
            report_independent(ReportKind::Violation, Region::Observed, b"gone\n".to_vec());
        });
        crate::sched::with_current_process(5, || {
            report_independent(ReportKind::Violation, Region::Observed, b"kept\n".to_vec());
        });
        PENDING.with(|p| {
            p.borrow_mut().insert(
                (ReportKind::DeferredImmediate, Some(4), 1),
                (99, b"other-kind\n".to_vec()),
            )
        });

        flush_violations(4);
        mature_all_pending();
        assert_eq!(captured(), "kept\nother-kind\n");

        reset_thread_locals();
    }

    /// LRM 20.11: `$assertkill` discards the queued reports of the assertions
    /// it reaches -- violation reports and deferred reports alike, whichever
    /// process queued them -- and nobody else's.
    #[test]
    fn flush_sites_discards_only_the_reports_of_the_killed_sites() {
        reset_thread_locals();
        CAPTURED.with(|c| c.borrow_mut().clear());
        set_writer(ReportKind::Violation, capture_writer);
        set_writer(ReportKind::DeferredImmediate, capture_writer);

        crate::sched::with_current_process(4, || {
            report_independent_for_site(
                ReportKind::Violation,
                Region::Observed,
                b"killed-violation\n".to_vec(),
                Some(7),
            );
            report_independent_for_site(
                ReportKind::Violation,
                Region::Observed,
                b"other-site\n".to_vec(),
                Some(8),
            );
            report_independent(
                ReportKind::Violation,
                Region::Observed,
                b"untagged\n".to_vec(),
            );
        });
        crate::sched::with_current_process(5, || {
            report_for_site(
                ReportKind::DeferredImmediate,
                11,
                Region::Reactive,
                b"killed-deferred\n".to_vec(),
                Some(7),
            );
        });
        flush_sites(&[7]);
        mature_all_pending();
        assert_eq!(captured(), "other-site\nuntagged\n");
        assert!(
            SITES.with(|s| s.borrow().is_empty()),
            "no tag outlives its report"
        );

        reset_thread_locals();
    }

    /// A tag follows its report out of the queue: a flush point or a newer
    /// report at the same key leaves no stale tag that a later kill could
    /// mistake for a pending one.
    #[test]
    fn site_tags_do_not_outlive_their_reports() {
        reset_thread_locals();
        crate::sched::with_current_process(4, || {
            report_independent_for_site(
                ReportKind::Violation,
                Region::Observed,
                b"x\n".to_vec(),
                Some(1),
            );
        });
        flush_violations(4);
        assert!(SITES.with(|s| s.borrow().is_empty()));
        crate::sched::with_current_process(4, || {
            report_for_site(
                ReportKind::DeferredImmediate,
                3,
                Region::Reactive,
                b"old\n".to_vec(),
                Some(2),
            );
            report_for_site(
                ReportKind::DeferredImmediate,
                3,
                Region::Reactive,
                b"new\n".to_vec(),
                None,
            );
        });
        flush_sites(&[2]);
        assert_eq!(
            PENDING.with(|p| p.borrow().len()),
            1,
            "the superseding untagged report is not site 2's"
        );
        reset_thread_locals();
    }

    /// The generic `flush` (an immediate-assertion flush point) leaves a
    /// process's violation reports alone: a delay or the process ending is not a
    /// violation flush point, so only `flush_violations` discards them.
    #[test]
    fn the_generic_flush_leaves_violation_reports_pending() {
        reset_thread_locals();
        CAPTURED.with(|c| c.borrow_mut().clear());
        set_writer(ReportKind::Violation, capture_writer);

        crate::sched::with_current_process(4, || {
            report_independent(
                ReportKind::Violation,
                Region::Observed,
                b"stands\n".to_vec(),
            );
        });
        flush(4);
        mature_all_pending();
        assert_eq!(captured(), "stands\n");

        reset_thread_locals();
    }
}