lazily 0.48.0

use std::any::{Any, TypeId};
use std::cell::RefCell;
use std::collections::{HashMap, VecDeque};
use std::rc::Rc;

#[cfg(not(feature = "vec_edges"))]
use smallvec::SmallVec;

use crate::cell::{Computed, Source};
use crate::effect::{Effect, EffectCallbackResult};
use crate::merge::MergePolicy;

/// Type alias for the erased compute function stored in slots.
type ComputeFn = dyn Fn(&Compute) -> AnyValue;
/// Type alias for the erased equality function stored in slots.
type EqualsFn = dyn Fn(&AnyValue, &AnyValue) -> bool;
/// Type alias for the erased effect callback stored in effects.
type EffectFn = dyn Fn(&Compute) -> Option<Box<dyn FnOnce()>>;

/// Unique identifier for a reactive node (slot or cell).
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)]
pub struct SlotId(pub(crate) u64);

impl SlotId {
    /// A sentinel id that never resolves to a live node. A [`Compute`] bound to
    /// it registers no edges — `register_dependency` no-ops on a missing node —
    /// so it is the id used for a one-shot *detached* eval outside any recompute
    /// (`Context::eval_detached`), where the unified `Fn(&Compute)` closure type
    /// must be honoured without forming spurious dependencies.
    pub(crate) const DETACHED: SlotId = SlotId(u64::MAX);
}

#[cfg(not(feature = "vec_edges"))]
type EdgeVec = SmallVec<[SlotId; 2]>;
#[cfg(feature = "vec_edges")]
type EdgeVec = Vec<SlotId>;

// The per-node `TypeId` exists only to power belt-and-suspenders type-mismatch
// asserts. In release builds it is a zero-sized `()` so it costs nothing to
// store (16 B/node saved at 10M-node scale); in debug builds it is a real
// `TypeId` and the asserts run.
#[cfg(debug_assertions)]
type TypeTag = TypeId;
#[cfg(not(debug_assertions))]
type TypeTag = ();

#[inline]
fn node_type_tag<T: 'static>() -> TypeTag {
    #[cfg(debug_assertions)]
    {
        TypeId::of::<T>()
    }
    #[cfg(not(debug_assertions))]
    {
        ()
    }
}

/// Degree at which an edge list stops scanning and gains a hash index.
///
/// #lzspecedgeindex. Dedup is a linear scan while a node's degree is small —
/// measurably faster than hashing at low degree, which is the overwhelmingly
/// common case — and gains a hash index above this threshold so a wide-fanout
/// node stays amortized O(1) per registration instead of degrading to O(n^2)
/// per propagation. An unconditional hash set regresses the common case; an
/// unconditional scan regresses wide fanout.
///
/// Measured crossover with `SlotIdHasher`: the indexed path costs ~45 ns per
/// registration flat, and a linear scan passes that near width 40. With `std`'s
/// SipHash the indexed path cost ~83 ns and the crossover sat near 170 — the
/// hasher, not the scan, was what made a low threshold look wrong here.
///
/// The index is held in a side table on `Inner`, not on the node, so a node
/// below the threshold carries no extra bytes at all. See `EdgeIndex`.
const EDGE_INDEX_THRESHOLD: usize = 32;

/// Default bound on effect-drain iterations within a single flush.
///
/// A scheduler-closed feedback loop — an effect that writes into its own
/// dependency cone — reschedules itself indefinitely. The write calls
/// `set_cell` -> `flush_effects`, which hits the re-entrancy guard and returns
/// immediately, appending the newly scheduled effect to the worklist for the
/// *outer* drain to pick up. So the loop runs flat, at constant stack depth: a
/// recursion-depth bound could never fire, and the drain's only exit is an
/// empty worklist. Without this budget a divergent loop hangs the process
/// rather than failing.
///
/// The value is deliberately large enough that a long-but-terminating cascade
/// does not trip it. Per-effect attribution (`DrainExhaustion::top_effects`) is
/// what distinguishes divergence from depth, so the budget does not have to be
/// tuned to tell them apart.
const DEFAULT_DRAIN_BUDGET: usize = 100_000;

/// Report produced when an effect drain exhausts its iteration budget.
///
/// Exhaustion is *not* convergence, and this deliberately does not pretend
/// otherwise: it reports that the drain was cut short, plus the effects that
/// ran most, so a diverging loop names itself instead of surfacing as "the
/// drain hit a counter".
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DrainExhaustion {
    /// Effect runs performed before the budget was reached.
    pub iterations: usize,
    /// The budget that was hit.
    pub budget: usize,
    /// `(effect id, run count)` for the busiest effects in the exhausted
    /// drain, descending by run count. A scheduler-closed loop puts its own
    /// effect at the head with a count proportional to `iterations`, which is
    /// what separates it from a wide cascade where runs are spread thin.
    pub top_effects: Vec<(u64, u32)>,
}

/// Hysteresis: demote only well below the promote threshold.
///
/// A dependent list oscillates by one on every recompute — edges are removed
/// and re-registered — so a single shared boundary makes a list sitting exactly
/// at the threshold demote and rebuild its index on every recompute. Measured
/// at ~4x the steady-state cost. The gap absorbs that oscillation.
const EDGE_INDEX_DEMOTE_THRESHOLD: usize = 24;

/// `owner -> (edge -> position in owner's edge list)`, for promoted nodes only.
///
/// Absent for every node below the threshold, which is why an unpromoted node
/// pays nothing: no field, no branch on the node itself, no allocation.
///
/// Entries MUST be dropped whenever the edge list they describe is cleared or
/// its owner is removed — `SlotId`s are recycled (`free_ids` is LIFO), so a
/// stale entry would silently alias a different node's edges.
/// Hasher for `SlotId` keys (#lzspecedgeindex).
///
/// `std`'s default is SipHash, chosen to resist collision attacks on
/// attacker-controlled keys. `SlotId`s are internally allocated sequential
/// integers that never come from outside the process, so that resistance buys
/// nothing here and is paid on every index lookup — twice per wide
/// registration, once for the owner and once for the edge.
///
/// This is the splitmix64 finalizer: full avalanche in a handful of
/// multiply-xor-shift ops. Sequential ids land in well-separated buckets.
#[derive(Default, Clone, Copy)]
pub(crate) struct SlotIdHasher(u64);

impl std::hash::Hasher for SlotIdHasher {
    fn finish(&self) -> u64 {
        self.0
    }

    fn write(&mut self, byte_a: &[u8]) {
        // SlotId hashes through write_u64; this exists only to satisfy the
        // trait, and is deliberately not tuned.
        for byte in byte_a {
            self.0 = (self.0 ^ u64::from(*byte)).wrapping_mul(0x100_0000_01b3);
        }
    }

    fn write_u64(&mut self, value: u64) {
        let mut mixed = value.wrapping_mul(0x9E37_79B9_7F4A_7C15);
        mixed = (mixed ^ (mixed >> 30)).wrapping_mul(0xBF58_476D_1CE4_E5B9);
        mixed = (mixed ^ (mixed >> 27)).wrapping_mul(0x94D0_49BB_1331_11EB);
        self.0 = mixed ^ (mixed >> 31);
    }
}

#[derive(Default, Clone)]
pub(crate) struct SlotIdHashBuilder;

impl std::hash::BuildHasher for SlotIdHashBuilder {
    type Hasher = SlotIdHasher;

    fn build_hasher(&self) -> Self::Hasher {
        SlotIdHasher(0)
    }
}

type OwnerEdgeIndex = HashMap<SlotId, usize, SlotIdHashBuilder>;
type EdgeIndex = HashMap<SlotId, OwnerEdgeIndex, SlotIdHashBuilder>;

/// Insert `id` into `edges` if absent. Returns whether an edge was added.
fn edge_insert(edges: &mut EdgeVec, id: SlotId, owner: SlotId, index: &mut EdgeIndex) -> bool {
    // Invariant: an owner has an index entry exactly while its edge list is
    // longer than the threshold. Gating on the length keeps a low-degree node
    // off the hash path entirely — hashing an absent key costs more than the
    // short scan it would replace.
    // `SmallVec::len` branches on inline-vs-spilled, so read it once: the
    // common path is then one compare on top of the scan it already did.
    let len = edges.len();
    // Below the demote threshold there is provably no index, so a short list
    // never touches the map. Between the thresholds one may or may not exist.
    if len > EDGE_INDEX_DEMOTE_THRESHOLD
        && let Some(owner_index) = index.get_mut(&owner)
    {
        if owner_index.contains_key(&id) {
            return false;
        }
        owner_index.insert(id, len);
        edges.push(id);
        return true;
    }
    if edges.contains(&id) {
        return false;
    }
    edges.push(id);
    if len + 1 > EDGE_INDEX_THRESHOLD && !index.contains_key(&owner) {
        // crossed the threshold on this push: build the index once
        index.insert(
            owner,
            edges
                .iter()
                .enumerate()
                .map(|(pos, edge)| (*edge, pos))
                .collect(),
        );
    }
    true
}

/// Remove `id` from `edges` if present. Returns whether an edge was removed.
///
/// Swap-removes, so edge order is not preserved — matching the previous
/// behaviour, which callers already tolerate.
fn edge_remove(edges: &mut EdgeVec, id: SlotId, owner: SlotId, index: &mut EdgeIndex) -> bool {
    if edges.len() > EDGE_INDEX_DEMOTE_THRESHOLD
        && let Some(owner_index) = index.get_mut(&owner)
    {
        // #lzspecedgeindex audit control: `--cfg naive_edge_remove` locates the
        // edge by linear scan instead of by stored position, keeping every
        // other step (including index maintenance) identical. Isolates the
        // scan, so a flat teardown column can be shown to be a real O(1)
        // removal rather than a harness that never touches this path.
        let pos = {
            #[cfg(naive_edge_remove)]
            {
                let scanned = edges.iter().position(|edge| *edge == id);
                owner_index.remove(&id);
                scanned
            }
            #[cfg(not(naive_edge_remove))]
            {
                owner_index.remove(&id)
            }
        };
        let Some(pos) = pos else {
            return false;
        };
        let last = edges
            .pop()
            .expect("index non-empty implies edges non-empty");
        if pos < edges.len() {
            edges[pos] = last;
            owner_index.insert(last, pos);
        }
        // Demote only well below the promote threshold, so a list hovering at
        // the boundary does not rebuild its index on every recompute.
        if edges.len() <= EDGE_INDEX_DEMOTE_THRESHOLD {
            index.remove(&owner);
        }
        return true;
    }
    if let Some(pos) = edges.iter().position(|edge| *edge == id) {
        edges.swap_remove(pos);
        true
    } else {
        false
    }
}

// ---------------------------------------------------------------------------
// SmallAny inline value storage (#lzsmallany)
// ---------------------------------------------------------------------------

// Inline storage envelope (mirrors the thread-safe INLINE_CAP/INLINE_ALIGN in
// thread_safe.rs): small, trivially-droppable values (i64/f64 and similar
// scalars) are stored inline in the node instead of behind a heap `Rc` box.
// lazily-cpp's SmallAny gives it a ~3x cold-recalc lead purely from skipping
// one heap allocation per recompute; this closes that gap.
const VALUE_INLINE_CAP: usize = 24;
const VALUE_INLINE_ALIGN: usize = 16;

// SAFETY: the `align(16)` guarantees any T with `align_of::<T>() <= 16` can be
// written into the buffer at its required alignment.
#[repr(C, align(16))]
pub(crate) struct InlineBuf([core::mem::MaybeUninit<u8>; VALUE_INLINE_CAP]);

/// Type-erased reactive value. `None` is the unset slot state; `Inline` holds a
/// small, trivially-droppable T bitwise (no `Rc` heap allocation); `Heap` falls
/// back to the classic `Rc<dyn Any>` for large or `Drop`-bearing types.
///
/// `Inline` is only ever used for T with `size_of::<T>() <= VALUE_INLINE_CAP`,
/// `align_of::<T>() <= VALUE_INLINE_ALIGN`, and `needs_drop::<T>() == false`,
/// so overwriting or dropping an `Inline` variant never needs to run a
/// destructor — the derived `Drop` only has to release the `Heap` `Rc`.
pub(crate) enum AnyValue {
    None,
    Inline(InlineBuf),
    Heap(Rc<dyn Any>),
}

impl AnyValue {
    /// Erase `value` into either inline storage or a heap `Rc`, picking inline
    /// only when it is safe to store bitwise (small, well-aligned, no drop).
    #[inline]
    pub(crate) fn from_value<T: 'static>(value: T) -> Self {
        if core::mem::size_of::<T>() <= VALUE_INLINE_CAP
            && core::mem::align_of::<T>() <= VALUE_INLINE_ALIGN
            && !core::mem::needs_drop::<T>()
        {
            let mut buf = InlineBuf([core::mem::MaybeUninit::uninit(); VALUE_INLINE_CAP]);
            // SAFETY: `buf` is 16-aligned (InlineBuf) and 24 bytes wide; T fits
            // both and is being written at its required alignment.
            unsafe {
                core::ptr::write(buf.0.as_mut_ptr() as *mut T, value);
            }
            AnyValue::Inline(buf)
        } else {
            AnyValue::Heap(Rc::new(value))
        }
    }

    /// Borrow the stored value as `&T`, trusting the caller's type (the node's
    /// `TypeTag` assert has already proven T matches). Works for both inline
    /// and heap storage.
    #[inline]
    pub(crate) unsafe fn as_t_ref_unchecked<T: 'static>(&self) -> &T {
        match self {
            AnyValue::Inline(buf) => unsafe { &*(buf.0.as_ptr() as *const T) },
            AnyValue::Heap(rc) => unsafe { &*(&**rc as *const dyn Any as *const T) },
            AnyValue::None => unsafe { core::hint::unreachable_unchecked() },
        }
    }

    /// Clone the stored value into a fresh `Rc<T>`. For heap storage this is a
    /// refcount bump (no deep clone); for inline storage there is no shared box
    /// to refcount, so a new `Rc` is materialized (inline-eligible T is trivially
    /// droppable, so owning it in an `Rc` is sound).
    #[inline]
    pub(crate) unsafe fn rc_clone_unchecked<T: 'static>(&self) -> Rc<T> {
        match self {
            AnyValue::Heap(rc) => {
                let rc: Rc<dyn Any> = Rc::clone(rc);
                unsafe {
                    let ptr = Rc::into_raw(rc) as *const T;
                    Rc::from_raw(ptr)
                }
            }
            AnyValue::Inline(buf) => {
                Rc::new(unsafe { core::ptr::read(buf.0.as_ptr() as *const T) })
            }
            AnyValue::None => unsafe { core::hint::unreachable_unchecked() },
        }
    }

    #[inline]
    pub(crate) fn is_none(&self) -> bool {
        matches!(self, AnyValue::None)
    }
}

// ---------------------------------------------------------------------------
// Dependency discovery (`#lzcellkernel`)
// ---------------------------------------------------------------------------
//
// Dependency tracking is threaded **entirely as a value**: the recomputing node
// id travels through the fortified [`Compute`] view handed to every
// compute/effect closure (see `ComputeOps` / `Compute`). A tracked read
// registers against `Compute::slot_id`; a bare `&Context` read registers
// nothing and is genuinely untracked. There is **no** ambient thread-local
// tracking frame — the whole class of stranded-frame bugs is structurally gone,
// and the last compatibility bridge (the `SyncReactiveGraph` capability trait's
// compute/effect closures) is retired now that the trait threads a value-carried
// `Self::Compute<'_>` view via a GAT (`src/reactive_graph.rs`).
//
// `ThreadSafeContext` / `AsyncContext` keep their *own* ambient engines; this
// removal is confined to the single-threaded `Context`.

// ---------------------------------------------------------------------------
// Internal node kinds stored inside Context
// ---------------------------------------------------------------------------

pub(crate) struct ComputedNode {
    pub(crate) value: AnyValue,
    pub(crate) type_id: TypeTag,
    pub(crate) compute: Rc<ComputeFn>,
    pub(crate) equals: Option<Box<EqualsFn>>,
    pub(crate) dependencies: EdgeVec,
    pub(crate) dependents: EdgeVec,
    pub(crate) dirty: bool,
    pub(crate) force_recompute: bool,
    /// True while this slot is actively refreshing/recomputing. Used to detect
    /// dependency cycles (a slot that reads itself directly or transitively)
    /// before the pull-based recompute walk overflows the stack.
    pub(crate) in_progress: bool,
    /// `#lzcellkernel` — whether this formula is **eager**: a puller
    /// effect keeps it materialized after every invalidation. Lands in the
    /// existing padding beside `dirty`/`force_recompute`/`in_progress`, so the
    /// node does not grow. The pulling effect's id lives in the `eager_by` side
    /// table, keyed by this slot's id — off the node, since eager formulas are
    /// rare (the `EdgeIndex` precedent).
    pub(crate) eager: bool,
    /// #lzspecrevisionengine: last global revision at which this slot was
    /// verified clean. In revision mode, staleness is `verified_at < revision`
    /// (O(1) write — no dirty walk) rather than the `dirty` flag.
    pub(crate) verified_at: u64,
}

pub(crate) struct SourceNode {
    pub(crate) value: AnyValue,
    pub(crate) type_id: TypeTag,
    pub(crate) dependents: EdgeVec,
}

pub(crate) struct EffectNode {
    /// The effect callback.
    pub(crate) run: Rc<EffectFn>,
    /// Slots/cells that this effect depends on. Populated during each run.
    pub(crate) dependencies: EdgeVec,
    /// Cleanup returned by the latest effect run, if any.
    pub(crate) cleanup: Option<Box<dyn FnOnce()>>,
    /// Whether this scheduled effect must run without dependency freshness checks.
    pub(crate) force_run: bool,
}

pub(crate) enum Node {
    Computed(ComputedNode),
    Source(SourceNode),
    Effect(EffectNode),
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
enum FactoryKind {
    Slot,
    Cell,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
struct FactoryKey {
    kind: FactoryKind,
    factory_type: TypeId,
}

struct FactoryEntry {
    value_type: TypeId,
    handle: Rc<dyn Any>,
}

// ---------------------------------------------------------------------------
// Context
// ---------------------------------------------------------------------------

struct ContextInner {
    nodes: Vec<Option<Node>>,
    next_id: u64,
    free_ids: Vec<u64>,
    /// #lzspecedgeindex. Hash indexes for promoted (wide) edge lists only; see
    /// `EdgeIndex`. Kept off the nodes so low-degree nodes cost nothing.
    dependents_index: EdgeIndex,
    dependencies_index: EdgeIndex,
    /// Reused buffer for invalidation roots. Swapped with a node's dependent
    /// list so a publish does not copy it — see `invalidate_dependents_now`.
    roots_scratch: EdgeVec,
    pending_effects: VecDeque<SlotId>,
    /// Effect-schedule membership bitset, indexed by node slot. Mirrors the
    /// thread-safe variant (thread_safe.rs): a `Vec<bool>` beats `HashSet` for
    /// the bounded, dense node-id space and avoids per-schedule hashing.
    scheduled_effects: Vec<bool>,
    flushing_effects: bool,
    /// Bound on effect runs within one drain. See `DEFAULT_DRAIN_BUDGET`.
    drain_budget: usize,
    /// Per-effect run counts for the drain currently in progress, indexed by
    /// node slot. Cleared lazily at the start of each outer drain so a
    /// terminating flush pays no scan.
    drain_runs: Vec<u32>,
    /// Set when a drain exhausts `drain_budget`. Observable rather than a
    /// panic so a conformance fixture can assert that a divergent loop reports
    /// exhaustion instead of hanging the runner.
    last_drain_exhaustion: Option<DrainExhaustion>,
    batch_depth: usize,
    batched_cells: EdgeVec,
    batched_cell_clears: EdgeVec,
    batched_slots: EdgeVec,
    /// Reusable DFS stack for `mark_frontier_locked` / `clear_frontier_locked`.
    /// Holds `(id, force)` so the former separate `stack` + `force_stack`
    /// allocations collapse into one with better pop locality (#lzbatchborrow).
    mark_scratch: Vec<(SlotId, bool)>,
    /// Reusable sink for `(effect_id, force)` pairs collected during a frontier
    /// walk. Taken out for one invalidation and restored afterward so its
    /// capacity survives across invalidations instead of reallocating per call.
    effects_scratch: Vec<(SlotId, bool)>,
    factory_handles: HashMap<FactoryKey, FactoryEntry>,
    /// #lzspecrevisionengine: global revision counter, bumped once per
    /// value-changing write. In revision mode, slot staleness is detected by
    /// `verified_at < revision` instead of the `dirty` flag, giving O(1)
    /// writes (no dependent cone walk). Push mode (default) leaves this at 0.
    revision: u64,
    /// #lzspecrevisionengine: whether this Context uses the revision (pull)
    /// invalidation engine instead of the default push (dirty-walk) engine.
    /// Per-Context choice; never mixed within one graph.
    revision_mode: bool,
    /// #lzspecrevisionengine: all effect node ids, for the revision-mode
    /// effect-flush scan (O(effects) per flush, not O(cone) per write).
    all_effect_ids: Vec<SlotId>,
    /// `#lzcellkernel` — owner-keyed side table mapping an **eager** formula's
    /// slot id to the id of the puller `Effect` that drives it. One entry per
    /// eager formula, zero per lazy one (the `EdgeIndex` precedent: per-node
    /// state for the common case, a side table for the rare one). MUST be
    /// cleared when a formula is disposed or made lazy — it is owner-keyed and
    /// would otherwise alias a recycled id (the `recycled_id_inherits_nothing`
    /// hazard).
    eager_by: HashMap<SlotId, SlotId>,
    #[cfg(feature = "instrumentation")]
    instrumentation: crate::instrumentation::InstrumentationCounters,
}

/// Container for all reactive nodes. Owns allocations; uses a single
/// interior-mutability cell (`RefCell`) for single-threaded use.
pub struct Context {
    inner: RefCell<ContextInner>,
}

struct BatchGuard<'a> {
    ctx: &'a Context,
}

impl Drop for BatchGuard<'_> {
    fn drop(&mut self) {
        self.ctx.finish_batch();
    }
}

/// RAII guard that clears a slot's `in_progress` cycle-detection flag when the
/// refresh of that slot completes (or unwinds).
struct RefreshGuard<'a> {
    ctx: &'a Context,
    id: SlotId,
}

impl Drop for RefreshGuard<'_> {
    fn drop(&mut self) {
        let mut inner = self.ctx.inner.borrow_mut();
        if let Some(Node::Computed(slot)) = Context::get_node_mut(&mut inner.nodes, self.id) {
            slot.in_progress = false;
        }
    }
}

impl Context {
    pub fn new() -> Self {
        Self::new_impl(false)
    }

    /// Create a Context using the **revision (pull) invalidation engine** instead
    /// of the default push (dirty-walk) engine (`#lzspecrevisionengine`).
    ///
    /// In revision mode, a cell write bumps a global revision counter (O(1),
    /// no dependent cone walk). Slot staleness is detected lazily on read via
    /// `verified_at < revision`. Observable values are provably identical to
    /// push mode (`get_equiv_push`, lazily-formal). Pick revision for
    /// write-heavy / high-fan-out workloads; keep push (default) for read-heavy.
    pub fn with_revision_engine() -> Self {
        Self::new_impl(true)
    }

    fn new_impl(revision_mode: bool) -> Self {
        Self {
            inner: RefCell::new(ContextInner {
                nodes: Vec::new(),
                next_id: 0,
                free_ids: Vec::new(),
                dependents_index: EdgeIndex::default(),
                dependencies_index: EdgeIndex::default(),
                roots_scratch: EdgeVec::new(),
                pending_effects: VecDeque::new(),
                drain_budget: DEFAULT_DRAIN_BUDGET,
                drain_runs: Vec::new(),
                last_drain_exhaustion: None,
                scheduled_effects: Vec::new(),
                flushing_effects: false,
                batch_depth: 0,
                batched_cells: EdgeVec::new(),
                batched_cell_clears: EdgeVec::new(),
                batched_slots: EdgeVec::new(),
                mark_scratch: Vec::new(),
                effects_scratch: Vec::new(),
                factory_handles: HashMap::new(),
                revision: 0,
                revision_mode,
                all_effect_ids: Vec::new(),
                eager_by: HashMap::new(),
                #[cfg(feature = "instrumentation")]
                instrumentation: crate::instrumentation::InstrumentationCounters::default(),
            }),
        }
    }

    pub(crate) fn alloc_id(&self) -> SlotId {
        let mut inner = self.inner.borrow_mut();
        let slot_id = match inner.free_ids.pop() {
            Some(id) => {
                let id = SlotId(id);
                // Belt and braces against id recycling: a fresh node must never
                // inherit an index entry. Disposal already clears these; this
                // makes any future removal path safe by construction.
                if !inner.dependents_index.is_empty() {
                    inner.dependents_index.remove(&id);
                }
                if !inner.dependencies_index.is_empty() {
                    inner.dependencies_index.remove(&id);
                }
                id
            }
            None => {
                let id = SlotId(inner.next_id);
                inner.next_id += 1;
                id
            }
        };
        #[cfg(feature = "instrumentation")]
        {
            inner.instrumentation.record_node_allocation();
        }
        slot_id
    }

    fn node_index(id: SlotId) -> Option<usize> {
        usize::try_from(id.0).ok()
    }

    fn get_node(nodes: &[Option<Node>], id: SlotId) -> Option<&Node> {
        nodes.get(Self::node_index(id)?)?.as_ref()
    }

    fn get_node_mut(nodes: &mut [Option<Node>], id: SlotId) -> Option<&mut Node> {
        nodes.get_mut(Self::node_index(id)?)?.as_mut()
    }

    fn take_node(nodes: &mut [Option<Node>], id: SlotId) -> Option<Node> {
        nodes.get_mut(Self::node_index(id)?)?.take()
    }

    fn insert_node(&self, id: SlotId, node: Node) {
        let index = Self::node_index(id).expect("SlotId does not fit usize");
        let mut inner = self.inner.borrow_mut();
        if inner.nodes.len() <= index {
            inner.nodes.resize_with(index + 1, || None);
        }
        inner.nodes[index] = Some(node);
    }

    pub(crate) fn register_dependency(&self, dependency_id: SlotId, dependent_id: SlotId) {
        if dependency_id == dependent_id {
            return;
        }

        #[cfg(feature = "instrumentation")]
        let mut edge_added = false;
        let mut inner = self.inner.borrow_mut();
        // Disjoint field borrows: the node comes from `nodes`, the index from
        // its own field, so both can be held at once.
        let inner_mut = &mut *inner;
        if let Some(node) = Self::get_node_mut(&mut inner_mut.nodes, dependency_id) {
            let index = &mut inner_mut.dependents_index;
            match node {
                Node::Computed(s) => {
                    edge_insert(&mut s.dependents, dependent_id, dependency_id, index);
                }
                Node::Source(c) => {
                    edge_insert(&mut c.dependents, dependent_id, dependency_id, index);
                }
                Node::Effect(_) => {}
            }
        }

        if let Some(node) = Self::get_node_mut(&mut inner_mut.nodes, dependent_id) {
            let index = &mut inner_mut.dependencies_index;
            match node {
                Node::Computed(parent) => {
                    #[cfg(feature = "instrumentation")]
                    {
                        edge_added = edge_insert(
                            &mut parent.dependencies,
                            dependency_id,
                            dependent_id,
                            index,
                        );
                    }
                    #[cfg(not(feature = "instrumentation"))]
                    {
                        edge_insert(&mut parent.dependencies, dependency_id, dependent_id, index);
                    }
                }
                Node::Effect(parent) => {
                    #[cfg(feature = "instrumentation")]
                    {
                        edge_added = edge_insert(
                            &mut parent.dependencies,
                            dependency_id,
                            dependent_id,
                            index,
                        );
                    }
                    #[cfg(not(feature = "instrumentation"))]
                    {
                        edge_insert(&mut parent.dependencies, dependency_id, dependent_id, index);
                    }
                }
                Node::Source(_) => {}
            }
        }

        #[cfg(feature = "instrumentation")]
        if edge_added {
            inner.instrumentation.record_dependency_edge_added();
        }
    }

    /// Remove every `dependent_id` edge from `old_deps`'s dependents under a
    /// SINGLE `ContextInner` borrow. Mirrors the thread-safe variant's
    /// `remove_stale_dependencies_locked` (thread_safe.rs): the former per-edge
    /// `remove_dependent_edge` calls each re-borrowed the `RefCell`, so a
    /// fan-out-256 recompute paid 256 borrow_mut acquisitions instead of one
    /// (#lzbatchborrow).
    /// Detach `dependency_id` from each of `dependent_a`'s dependency lists.
    ///
    /// The mirror of `remove_dependent_edges_locked`, needed when a node is
    /// torn down while others still point at it.
    fn remove_dependency_edges_locked(
        inner: &mut ContextInner,
        dependency_id: SlotId,
        dependent_a: &[SlotId],
    ) {
        for dependent_id in dependent_a {
            let inner_mut = &mut *inner;
            if let Some(node) = Self::get_node_mut(&mut inner_mut.nodes, *dependent_id) {
                let index = &mut inner_mut.dependencies_index;
                match node {
                    Node::Computed(slot) => {
                        edge_remove(&mut slot.dependencies, dependency_id, *dependent_id, index);
                    }
                    Node::Effect(effect) => {
                        edge_remove(
                            &mut effect.dependencies,
                            dependency_id,
                            *dependent_id,
                            index,
                        );
                    }
                    Node::Source(_) => {}
                }
            }
        }
    }

    /// Drop both index side-table entries for `id`.
    ///
    /// Mandatory before recycling an id: `free_ids` is LIFO, so a stale entry
    /// would alias the very next node allocated.
    fn drop_edge_index_entries(inner: &mut ContextInner, id: SlotId) {
        if !inner.dependencies_index.is_empty() {
            inner.dependencies_index.remove(&id);
        }
        if !inner.dependents_index.is_empty() {
            inner.dependents_index.remove(&id);
        }
    }

    fn remove_dependent_edges_locked(
        inner: &mut ContextInner,
        dependent_id: SlotId,
        old_deps: &[SlotId],
    ) {
        for dependency_id in old_deps {
            #[cfg(feature = "instrumentation")]
            let mut edge_removed = false;
            let inner_mut = &mut *inner;
            if let Some(dep_node) = Self::get_node_mut(&mut inner_mut.nodes, *dependency_id) {
                let index = &mut inner_mut.dependents_index;
                match dep_node {
                    Node::Computed(s) => {
                        #[cfg(feature = "instrumentation")]
                        {
                            edge_removed =
                                edge_remove(&mut s.dependents, dependent_id, *dependency_id, index);
                        }
                        #[cfg(not(feature = "instrumentation"))]
                        {
                            edge_remove(&mut s.dependents, dependent_id, *dependency_id, index);
                        }
                    }
                    Node::Source(c) => {
                        #[cfg(feature = "instrumentation")]
                        {
                            edge_removed =
                                edge_remove(&mut c.dependents, dependent_id, *dependency_id, index);
                        }
                        #[cfg(not(feature = "instrumentation"))]
                        {
                            edge_remove(&mut c.dependents, dependent_id, *dependency_id, index);
                        }
                    }
                    Node::Effect(_) => {}
                }
            }
            #[cfg(feature = "instrumentation")]
            if edge_removed {
                inner.instrumentation.record_dependency_edge_removed();
            }
        }
    }

    // -- Slot API ----------------------------------------------------------

    /// Create a **pass-through** computed cell — a derived value with **no
    /// guard** (`#lzcellkernel`).
    ///
    /// Every recompute propagates downstream (no equality suppression). Use it
    /// for a `T` that has no cheap equality and that you're fine re-firing (a
    /// value without `PartialEq`, or one where a compare would cost more than a
    /// downstream recompute). When you *do* want change-suppression, prefer
    /// [`computed`](Self::computed) (natural equality) or
    /// [`computed_ripple_when`](Self::computed_ripple_when) (custom change predicate).
    /// `T` needs no `PartialEq` bound, which is why this is the escape for
    /// non-equatable derived values.
    pub fn slot<T, F>(&self, compute: F) -> Computed<T>
    where
        T: 'static,
        F: Fn(&Compute) -> T + 'static,
    {
        self.slot_with_equals(compute, None)
    }

    /// Create a **computed cell** — a value computed from upstream, returned as
    /// a [`Computed`] handle. This is the primary derived constructor.
    ///
    /// **Guarded by default** (`#lzcellkernel`): every computed cell carries a
    /// `PartialEq` equality guard, so a recompute that produces a value equal to
    /// the cached one suppresses downstream invalidation (the version is not
    /// bumped) — matching TC39 `Signal.Computed`. There is no unguarded mode;
    /// the former `memo` constructor is retired because `computed` now *is* the
    /// guarded form. Use [`signal`](Context::signal) for the eager (eager,
    /// guarded) construction. Lazy until
    /// [`Computed::eager`](crate::Computed::eager) makes it eager.
    pub fn computed<T, F>(&self, compute: F) -> Computed<T>
    where
        T: PartialEq + 'static,
        F: Fn(&Compute) -> T + 'static,
    {
        self.slot_with_equals(
            compute,
            Some(Box::new(|old: &AnyValue, new: &AnyValue| {
                let old = unsafe { old.as_t_ref_unchecked::<T>() };
                let new = unsafe { new.as_t_ref_unchecked::<T>() };
                old == new
            })),
        )
    }

    /// Create a **guarded computed cell** with an explicit change predicate
    /// (`#lzcellkernel`).
    ///
    /// Like [`computed`](Self::computed), but propagation is gated by
    /// `changed(old, new)` instead of the value's natural `PartialEq`:
    /// `changed` returns `true` to **propagate** the recompute downstream, and
    /// `false` to **suppress** it (treat it as "no meaningful change"). So
    /// `computed(f)` is exactly `computed_ripple_when(f, |o, n| o != n)`, and
    /// [`slot`](Self::slot) is `computed_ripple_when(f, |_, _| true)` (always
    /// propagate).
    ///
    /// Use it for a **custom significance policy** — dedup a large value by a
    /// version/hash field, epsilon float compare, hysteresis, a monotonic gate,
    /// or "propagate every N" when the counter lives in the value — or for a
    /// `T` that has no `PartialEq`.
    ///
    /// The value is **always computed** (the predicate needs `new`); `when`
    /// gates only the downstream cascade, not the computation. `changed` MUST
    /// be a pure function of `(old, new)` — reading value-carried state
    /// (version/counter/sequence) is fine and stays deterministic; capturing
    /// external mutable state is not (it keys off recompute/read frequency
    /// under laziness and breaks determinism).
    pub fn computed_ripple_when<T, F, C>(&self, compute: F, changed: C) -> Computed<T>
    where
        T: 'static,
        F: Fn(&Compute) -> T + 'static,
        C: Fn(&T, &T) -> bool + 'static,
    {
        self.slot_with_equals(
            compute,
            // Internal engine guards on equality (true = equal = suppress);
            // `changed` is its negation (true = propagate).
            Some(Box::new(move |old: &AnyValue, new: &AnyValue| {
                let old = unsafe { old.as_t_ref_unchecked::<T>() };
                let new = unsafe { new.as_t_ref_unchecked::<T>() };
                !changed(old, new)
            })),
        )
    }

    /// Return the context-local slot handle for factory `K`, creating it on
    /// first use.
    ///
    /// This supports decorator-style factory functions: callers do not store
    /// handles in wrapper structs; the context memoizes one handle per factory
    /// key. Later calls with the same key return the same slot handle and ignore
    /// the supplied compute callback.
    pub fn memoized_slot<K, T, F>(&self, compute: F) -> Computed<T>
    where
        K: 'static,
        T: 'static,
        F: Fn(&Compute) -> T + 'static,
    {
        let key = FactoryKey {
            kind: FactoryKind::Slot,
            factory_type: TypeId::of::<K>(),
        };
        if let Some(handle) = self.factory_handle::<Computed<T>>(key, TypeId::of::<T>()) {
            return handle;
        }

        let handle = self.slot(compute);
        self.insert_factory_handle(key, TypeId::of::<T>(), handle);
        handle
    }

    fn slot_with_equals<T, F>(&self, compute: F, equals: Option<Box<EqualsFn>>) -> Computed<T>
    where
        T: 'static,
        F: Fn(&Compute) -> T + 'static,
    {
        let id = self.alloc_id();
        let node = ComputedNode {
            value: AnyValue::None,
            type_id: node_type_tag::<T>(),
            compute: Rc::new(move |ctx| AnyValue::from_value(compute(ctx))),
            equals,
            dependencies: EdgeVec::new(),
            dependents: EdgeVec::new(),
            dirty: false,
            force_recompute: false,
            in_progress: false,
            eager: false,
            verified_at: 0,
        };
        self.insert_node(id, Node::Computed(node));
        Computed::from_id(id)
    }

    /// Read the current value of a cell — a [`Computed`] or a [`Source`].
    ///
    /// The unified read of the Cell kernel (`#lzcellkernel`). Supersedes the
    /// deprecated `get_cell`; a `Computed` recomputes if necessary, a `Source`
    /// returns its stored value.
    pub fn get<H: Read<Self> + ?Sized>(&self, handle: &H) -> <H as Read<Self>>::Output {
        handle.read(self)
    }

    /// Write a new value to a [`Source`] cell.
    ///
    /// The unified write of the Cell kernel (`#lzcellkernel`). Supersedes the
    /// deprecated `set_cell`. Only `Source` implements [`ContextWrite`], so
    /// `ctx.set(computed, v)` is a compile error (write protection, design §3).
    pub fn set<H: Write<Self> + ?Sized>(&self, handle: &H, value: <H as Write<Self>>::Value) {
        handle.write(self, value)
    }

    /// Get the value of a slot as `Rc<T>`, avoiding a deep clone.
    ///
    /// Returns a reference-counted pointer to the stored value. Use this when
    /// you only need to read the value without owning a separate copy.
    pub fn get_rc<T: 'static>(&self, handle: &Computed<T>) -> Rc<T> {
        // A bare `&Context` read is untracked; tracked reads go through
        // `Compute::get_rc` (value-threaded), which registers its own edge.
        self.get_rc_untracked(handle)
    }

    /// [`get_rc`](Self::get_rc) without registering a dependency edge — the
    /// fortified [`Compute`] wrapper registers against its own node instead.
    fn get_rc_untracked<T: 'static>(&self, handle: &Computed<T>) -> Rc<T> {
        self.refresh_slot(handle.id);

        let inner = self.inner.borrow();
        if let Some(Node::Computed(slot)) = Self::get_node(&inner.nodes, handle.id)
            && !slot.value.is_none()
        {
            assert!(
                slot.type_id == node_type_tag::<T>(),
                "type mismatch in slot"
            );
            return unsafe { slot.value.rc_clone_unchecked::<T>() };
        }
        panic!("get_rc called on unset or non-slot id");
    }

    /// Read a source cell's value **without** registering a dependency edge.
    fn read_source_untracked<T: Clone + 'static>(&self, id: SlotId) -> T {
        let inner = self.inner.borrow();
        if let Some(Node::Source(c)) = Self::get_node(&inner.nodes, id) {
            assert!(c.type_id == node_type_tag::<T>(), "type mismatch in cell");
            unsafe { c.value.as_t_ref_unchecked::<T>() }.clone()
        } else {
            panic!("get called on non-cell id");
        }
    }

    /// Internal: get a slot value by id, performing computation if unset and
    /// registering dependency tracking.
    fn get_slot<T: Clone + 'static>(&self, id: SlotId) -> T {
        // A bare `&Context` read is untracked; the tracking path is `Compute`.
        self.get_slot_untracked(id)
    }

    /// Read a computed slot's value **without** registering a dependency edge.
    /// The fortified `Compute` wrapper (`#lzcellkernel`) uses this and registers
    /// the edge against its own value-threaded node id instead of the ambient
    /// tracking frame.
    fn get_slot_untracked<T: Clone + 'static>(&self, id: SlotId) -> T {
        self.refresh_slot(id);

        let inner = self.inner.borrow();
        if let Some(Node::Computed(slot)) = Self::get_node(&inner.nodes, id)
            && !slot.value.is_none()
        {
            assert!(
                slot.type_id == node_type_tag::<T>(),
                "type mismatch in slot"
            );
            return unsafe { slot.value.as_t_ref_unchecked::<T>() }.clone();
        }
        panic!("get_slot called on unset or non-slot id");
    }

    /// Refresh a slot if its cached value may be stale.
    ///
    /// Returns true only when the slot's computed value changed. Downstream
    /// dependents use this as the memoization guard: a dirty dependency whose
    /// value recomputes equal does not force them to recompute.
    /// Mark a slot as actively refreshing, returning a RAII guard that clears
    /// the flag on drop. Returns `None` for non-slot ids (nothing to refresh).
    ///
    /// Panics if the slot is already in-progress: that means the recompute walk
    /// has re-entered a slot still on the call stack, i.e. a dependency cycle.
    /// Surfacing it as a deterministic panic turns an otherwise-divergent
    /// infinite recompute / stack overflow into a recoverable error a caller
    /// can `catch_unwind` and render as a `#CIRCULAR!`-style value.
    fn enter_refresh(&self, id: SlotId) -> Option<RefreshGuard<'_>> {
        let mut inner = self.inner.borrow_mut();
        match Self::get_node_mut(&mut inner.nodes, id) {
            Some(Node::Computed(slot)) => {
                if slot.in_progress {
                    drop(inner);
                    panic!(
                        "lazily: circular dependency detected at slot {id:?}; a \
                         computed/memo slot depends on itself (directly or \
                         transitively) and would recompute infinitely. Break the \
                         cycle (e.g. via a base case or an untracked read)."
                    );
                }
                slot.in_progress = true;
                Some(RefreshGuard { ctx: self, id })
            }
            _ => None,
        }
    }

    fn refresh_slot(&self, id: SlotId) -> bool {
        // Fast path: clean cache hit. When the slot holds a value and is
        // neither dirty nor force-recompute, no upstream value can have
        // changed since the last compute — invalidation always sets
        // `dirty=true` on dependents via `mark_slot_dirty` (called from
        // `invalidate_dependent_from_changed_value` with `force_recompute=true`
        // for both cell- and slot-driven changes). The dependency-refresh walk,
        // the cycle guard, and the dirty-flag clear are therefore all
        // unnecessary on this path. This is the hot path for cached slot
        // reads: it collapses the borrowMut (enter_refresh) + guard-drop
        // borrowMut + dependencies Vec clone + per-dep `is_slot_node` borrows
        // + needs_recompute borrow + clear_slot_dirty_flags borrowMut down to a
        // single shared borrow.
        {
            let inner = self.inner.borrow();
            match Self::get_node(&inner.nodes, id) {
                Some(Node::Computed(slot)) => {
                    if inner.revision_mode {
                        // #lzspecrevisionengine: staleness = verified_at < revision.
                        // Clean (verified_at == revision) and has a value → cache hit.
                        if !slot.value.is_none() && slot.verified_at == inner.revision {
                            return false;
                        }
                    } else if !slot.value.is_none() && !slot.dirty && !slot.force_recompute {
                        return false;
                    }
                }
                _ => return false,
            }
        }

        // Cycle guard: mark this slot in-progress for the duration of the
        // refresh. If the pull-based walk re-enters the same slot (a slot that
        // depends on itself directly or transitively), `enter_refresh` panics
        // with a diagnostic instead of recursing into a stack overflow.
        let Some(_cycle_guard) = self.enter_refresh(id) else {
            return false;
        };

        let dependencies = {
            let inner = self.inner.borrow();
            match Self::get_node(&inner.nodes, id) {
                Some(Node::Computed(slot)) => slot.dependencies.clone(),
                _ => return false,
            }
        };

        let mut dependency_changed = false;
        for dep_id in dependencies {
            // refresh_slot returns false for non-slot ids (its fast path and
            // every later borrow fall through to `_ => ...`), so the explicit
            // is_slot_node borrow+lookup is redundant — calling refresh_slot
            // directly collapses two RefCell borrows per dependency into one.
            if self.refresh_slot(dep_id) {
                dependency_changed = true;
            }
        }

        let needs_recompute = {
            let inner = self.inner.borrow();
            let slot = match Self::get_node(&inner.nodes, id) {
                Some(Node::Computed(slot)) => slot,
                _ => return false,
            };
            if inner.revision_mode {
                // #lzspecrevisionengine: reaching here means verified_at <
                // revision (the fast path didn't short-circuit). The slot is
                // stale and must recompute. The memo guard in recompute_slot_now
                // handles the value early-cutoff (if the recomputed value
                // equals the cache, downstream caches are preserved).
                true
            } else {
                slot.value.is_none() || slot.force_recompute || dependency_changed
            }
        };

        if !needs_recompute {
            self.clear_slot_dirty_flags(id);
            return false;
        }

        self.recompute_slot_now(id)
    }

    fn is_slot_node(&self, id: SlotId) -> bool {
        let inner = self.inner.borrow();
        matches!(Self::get_node(&inner.nodes, id), Some(Node::Computed(_)))
    }

    fn clear_slot_dirty_flags(&self, id: SlotId) {
        let mut inner = self.inner.borrow_mut();
        if let Some(Node::Computed(slot)) = Self::get_node_mut(&mut inner.nodes, id) {
            slot.dirty = false;
            slot.force_recompute = false;
        }
    }

    fn recompute_slot_now(&self, id: SlotId) -> bool {
        let compute: Rc<ComputeFn>;
        let old_deps;
        {
            let mut inner = self.inner.borrow_mut();
            #[cfg(feature = "instrumentation")]
            {
                inner.instrumentation.record_slot_recompute();
            }
            let slot = match Self::get_node_mut(&mut inner.nodes, id) {
                Some(Node::Computed(s)) => s,
                _ => panic!("get_slot called on non-slot id"),
            };
            old_deps = std::mem::take(&mut slot.dependencies);
            compute = Rc::clone(&slot.compute);
            // The list this described is gone; its index must go with it.
            // `is_empty` first: with no promoted node anywhere this is a
            // pointer compare, where `remove` would hash on every recompute.
            if !inner.dependencies_index.is_empty() {
                inner.dependencies_index.remove(&id);
            }
            Self::remove_dependent_edges_locked(&mut inner, id, &old_deps);
        }

        let result = {
            let cx = Compute::new(self, id, 0);
            (compute.as_ref())(&cx)
        };

        let changed = {
            let mut inner = self.inner.borrow_mut();
            let (rev_mode, rev) = (inner.revision_mode, inner.revision);
            let slot = match Self::get_node_mut(&mut inner.nodes, id) {
                Some(Node::Computed(slot)) => slot,
                _ => return false,
            };
            let had_value = !slot.value.is_none();
            let unchanged = match (&slot.value, &slot.equals) {
                (AnyValue::None, _) | (_, None) => false,
                (old, Some(equals)) => equals(old, &result),
            };
            slot.dirty = false;
            slot.force_recompute = false;
            if rev_mode {
                slot.verified_at = rev;
            }
            if unchanged {
                false
            } else {
                slot.value = result;
                had_value
            }
        };

        if changed {
            self.notify_slot_value_changed(id);
        }

        changed
    }

    /// Get the value of a cell.
    #[deprecated(note = "use `Context::get` — the unified cell read (#lzcellkernel)")]
    pub fn get_cell<T: Clone + 'static>(&self, handle: &Source<T>) -> T {
        self.get(handle)
    }

    /// Get the value of a cell as `Rc<T>`, avoiding a deep clone.
    pub fn get_cell_rc<T: 'static>(&self, handle: &Source<T>) -> Rc<T> {
        // A bare `&Context` read is untracked; the tracking path is
        // `Compute::get_cell_rc`, which registers its own edge.
        let inner = self.inner.borrow();
        if let Some(Node::Source(c)) = Self::get_node(&inner.nodes, handle.id) {
            assert!(c.type_id == node_type_tag::<T>(), "type mismatch in cell");
            unsafe { c.value.rc_clone_unchecked::<T>() }
        } else {
            panic!("get_cell_rc called on non-cell id");
        }
    }

    /// `#lzcellkernel` — read any node's value by id, dispatching on its kind:
    /// a formula (Slot node) is refreshed then read; a source (Cell node) is
    /// read directly. Registers a dependency when called inside a reactive
    /// computation. Backs `Source::get` / `Computed::get` uniformly across kinds.
    pub(crate) fn read_value<T: Clone + 'static>(&self, id: SlotId) -> T {
        // A bare `&Context` read is untracked; the primary path tracks via
        // `Compute::read_value`, which registers against its threaded node id and
        // calls the untracked base.
        self.read_value_untracked(id)
    }

    /// Read any node's value by id **without** registering a dependency edge.
    /// A formula (Computed node) is refreshed then read; a source (Source node)
    /// is read directly. The untracked base of [`read_value`](Self::read_value)
    /// used by the fortified [`Compute`] view (which registers its own edge).
    pub(crate) fn read_value_untracked<T: Clone + 'static>(&self, id: SlotId) -> T {
        // No-op for non-slot nodes (`refresh_slot` early-returns `false`).
        self.refresh_slot(id);
        let inner = self.inner.borrow();
        match Self::get_node(&inner.nodes, id) {
            Some(Node::Computed(slot)) if !slot.value.is_none() => {
                assert!(
                    slot.type_id == node_type_tag::<T>(),
                    "type mismatch in read"
                );
                unsafe { slot.value.as_t_ref_unchecked::<T>() }.clone()
            }
            Some(Node::Source(c)) => {
                assert!(c.type_id == node_type_tag::<T>(), "type mismatch in read");
                unsafe { c.value.as_t_ref_unchecked::<T>() }.clone()
            }
            _ => panic!("read_value called on unset or unknown id"),
        }
    }

    /// Evaluate `f` once, **detached** from any recompute: reads through the
    /// supplied [`Compute`] register no dependency edge (it is bound to
    /// [`SlotId::DETACHED`]). Used to produce a value under the unified
    /// `Fn(&Compute)` closure type when there is no owning node yet — e.g. a
    /// source cell's eager initial value in the keyed-collection family.
    pub(crate) fn eval_detached<T>(&self, f: impl FnOnce(&Compute) -> T) -> T {
        f(&Compute::new(self, SlotId::DETACHED, 0))
    }

    // -- Cell kernel constructors (`#lzcellkernel`) ------------------------

    /// Create a **source cell** — a value written from outside under the default
    /// [`KeepLatest`](crate::KeepLatest) policy (last-writer-wins replace). This
    /// is the old `cell(v)`. For another merge policy use
    /// [`source_with`](Context::source_with). Only a `Source` has `set`/`merge`.
    ///
    /// (The design's single `source::<M>(v)` form is not expressible in Rust —
    /// functions cannot carry a default type parameter — so the policy variant is
    /// the separate `source_with::<M>(v)`.)
    pub fn source<T>(&self, value: T) -> Source<T>
    where
        T: PartialEq + 'static,
    {
        self.cell(value)
    }

    /// Create a **source cell** whose write folds under merge policy `M`
    /// (`source_with::<Sum>(v)` is the old `merge_cell::<Sum>(v)`). A
    /// `Source<T, M>` with `M != KeepLatest`.
    pub fn source_with<M, T>(&self, value: T) -> Source<T, M>
    where
        T: PartialEq + 'static,
        M: MergePolicy<T>,
    {
        Source::from_id(self.cell(value).id)
    }

    /// `#lzcellkernel` — make a formula eager by id. Idempotent: a
    /// no-op if the formula is already eager. Attaches a puller effect that
    /// re-materializes the formula after every invalidation and records it in the
    /// `eager_by` side table. Non-formula ids are ignored.
    pub(crate) fn make_eager<T: 'static>(&self, id: SlotId) {
        {
            let inner = self.inner.borrow();
            match Self::get_node(&inner.nodes, id) {
                Some(Node::Computed(slot)) if slot.eager => return,
                Some(Node::Computed(_)) => {}
                _ => return,
            }
        }
        // Eager puller: re-materializes the formula after every invalidation.
        // `get_rc` refreshes and registers the dependency without deep-cloning.
        let effect = self.effect(move |ctx| {
            let _ = ctx.get_rc::<T>(&Computed::<T>::from_id(id));
        });
        let mut inner = self.inner.borrow_mut();
        if let Some(Node::Computed(slot)) = Self::get_node_mut(&mut inner.nodes, id) {
            slot.eager = true;
        }
        inner.eager_by.insert(id, effect.id);
    }

    /// `#lzcellkernel` — make a formula lazy (revert from eager): dispose its
    /// puller effect, clear the eager bit, and remove its `eager_by` entry. No-op
    /// if not eager. This is the reverse transition that replaces `dispose_signal`.
    pub(crate) fn make_lazy(&self, id: SlotId) {
        let effect_id = {
            let mut inner = self.inner.borrow_mut();
            match Self::get_node_mut(&mut inner.nodes, id) {
                Some(Node::Computed(slot)) if slot.eager => slot.eager = false,
                _ => return,
            }
            inner.eager_by.remove(&id)
        };
        if let Some(effect_id) = effect_id {
            self.dispose_effect(&Effect::new(effect_id));
        }
    }

    /// `#lzcellkernel` — whether the formula at `id` is currently eager.
    pub(crate) fn is_eager(&self, id: SlotId) -> bool {
        let inner = self.inner.borrow();
        matches!(
            Self::get_node(&inner.nodes, id),
            Some(Node::Computed(slot)) if slot.eager
        )
    }

    /// `#lzcellkernel` — tear down a node by id, dispatching on its own kind.
    /// Backs the kind-agnostic `Source::dispose` / `Computed::dispose`.
    pub(crate) fn dispose_node(&self, id: SlotId) {
        self.dispose_id(id);
    }

    // -- Cell API ----------------------------------------------------------

    /// Create a new mutable cell with an initial value.
    pub fn cell<T: PartialEq + 'static>(&self, value: T) -> Source<T> {
        let id = self.alloc_id();
        let node = SourceNode {
            value: AnyValue::from_value(value),
            type_id: node_type_tag::<T>(),
            dependents: EdgeVec::new(),
        };
        self.insert_node(id, Node::Source(node));
        Source::from_id(id)
    }

    /// Create a [`Source`] — a cell whose write is a *merge* under
    /// policy `M`, rather than a replace. `Cell ≡ MergeCell<KeepLatest>`
    /// (relaycell-backpressure-analysis.md §4.0). Backed by an ordinary cell
    /// node, so it inherits the store-without-cascade write fast path.
    pub fn merge_cell<T, M>(&self, initial: T) -> Source<T, M>
    where
        T: PartialEq + 'static,
        M: MergePolicy<T>,
    {
        // A merge cell is just a `Source<T, M>` over an ordinary cell node —
        // same node as a plain cell, the policy lives in `Source<T, M>`'s param.
        Source::from_id(self.cell(initial).id)
    }

    /// Fold `op` into a cell's value under policy `M` (the merge write). Reads
    /// the current value untracked, computes `M::merge(old, op)`, then routes
    /// through [`set_cell`](Context::set_cell) so the `PartialEq` store-guard
    /// (free dedup when `⊕(old, op) == old`), batching, and
    /// store-without-cascade all apply unchanged.
    pub fn apply_merge<T, M>(&self, handle: &Source<T>, op: T)
    where
        T: PartialEq + Clone + 'static,
        M: MergePolicy<T>,
    {
        self.merge_source::<T, M>(handle.id, op);
    }

    /// `#lzcellkernel` — the merge write by node id, backing `Source::merge`
    /// and `Context::apply_merge`. Reads the current value untracked, folds
    /// `M::merge(old, op)`, then routes through [`set_source`](Context::set_source)
    /// so the `PartialEq` store-guard, batching, and store-without-cascade all
    /// apply unchanged.
    pub(crate) fn merge_source<T, M>(&self, id: SlotId, op: T)
    where
        T: PartialEq + Clone + 'static,
        M: MergePolicy<T>,
    {
        let merged = {
            let inner = self.inner.borrow();
            if let Some(Node::Source(c)) = Self::get_node(&inner.nodes, id) {
                assert!(
                    c.type_id == node_type_tag::<T>(),
                    "type mismatch in apply_merge"
                );
                let old = unsafe { c.value.as_t_ref_unchecked::<T>() };
                M::merge(old, op)
            } else {
                panic!("apply_merge on non-cell id");
            }
        };
        self.set_source::<T>(id, merged);
    }

    /// Return the context-local cell handle for factory `K`, creating it on
    /// first use.
    ///
    /// The initializer belongs to the factory. It runs only when this context
    /// has not seen `K` before; callers should mutate the returned cell handle
    /// with [`Source::set`] / [`Context::set_cell`].
    pub fn memoized_cell<K, T, F>(&self, init: F) -> Source<T>
    where
        K: 'static,
        T: PartialEq + 'static,
        F: FnOnce(&Compute) -> T,
    {
        let key = FactoryKey {
            kind: FactoryKind::Cell,
            factory_type: TypeId::of::<K>(),
        };
        if let Some(handle) = self.factory_handle::<Source<T>>(key, TypeId::of::<T>()) {
            return handle;
        }

        // A memoized *cell* seed is a one-shot untracked eval (`#lzcellkernel`).
        let value = self.eval_detached(init);
        let handle = self.cell(value);
        self.insert_factory_handle(key, TypeId::of::<T>(), handle);
        handle
    }

    /// Set the value of a cell. If the value differs (via PartialEq),
    /// dependent slots are marked dirty for memoized validation.
    #[deprecated(note = "use `Context::set` — the unified cell write (#lzcellkernel)")]
    pub fn set_cell<T: PartialEq + 'static>(&self, handle: &Source<T>, new_value: T) {
        self.set(handle, new_value);
    }

    /// `#lzcellkernel` — the source write by node id, backing `Source::set`
    /// and `Context::set_cell`.
    pub(crate) fn set_source<T: PartialEq + 'static>(&self, id: SlotId, new_value: T) {
        let handle_id = id;
        let changed = {
            let inner = self.inner.borrow();
            if let Some(Node::Source(c)) = Self::get_node(&inner.nodes, handle_id) {
                assert!(
                    c.type_id == node_type_tag::<T>(),
                    "type mismatch in cell set"
                );
                let old = unsafe { c.value.as_t_ref_unchecked::<T>() };
                *old != new_value
            } else {
                panic!("set_cell on non-cell id");
            }
        };

        if changed {
            {
                let mut inner = self.inner.borrow_mut();
                if let Some(Node::Source(c)) = Self::get_node_mut(&mut inner.nodes, handle_id) {
                    c.value = AnyValue::from_value(new_value);
                }
            }
            if self.is_batching() {
                self.inner.borrow_mut().batched_cells.push(handle_id);
            } else if self.inner.borrow().revision_mode {
                // #lzspecrevisionengine: O(1) write — bump the global revision
                // counter; no dependent cone walk. Slot staleness is detected
                // lazily on read via `verified_at < revision`. Effects are
                // notified via the revision-mode flush scan.
                self.inner.borrow_mut().revision += 1;
                self.flush_effects_revision();
            } else {
                // Store-without-cascade: dirty-mark the dependent cone, then flush
                // effects ONLY when the cone actually contains an Effect. A cell
                // with no active (Effect-bearing) dependent stores its latest
                // value (already done above, so a late subscriber reads it
                // glitch-free) and marks lazy Slot dependents dirty, but pays no
                // effect-scheduling flush — the write side of the merge cost law
                // (relaycell-backpressure-analysis.md §4.0 / §5).
                if self.invalidate_cell_dependents_now(handle_id) {
                    self.flush_effects();
                }
            }
        }
    }

    // -- Batch API ---------------------------------------------------------

    /// Run several updates as one invalidation pass.
    ///
    /// Cell updates and explicit clears inside the callback are collected and
    /// applied when the outermost batch completes. Direct cell reads see the
    /// latest values immediately; dependent slots keep their previous cached
    /// values until the batch exits.
    pub fn batch<F, R>(&self, run: F) -> R
    where
        F: FnOnce(&Context) -> R,
    {
        self.inner.borrow_mut().batch_depth += 1;
        let _guard = BatchGuard { ctx: self };
        run(self)
    }

    fn finish_batch(&self) {
        let should_flush = {
            let mut inner = self.inner.borrow_mut();
            assert!(
                inner.batch_depth > 0,
                "finish_batch called without active batch"
            );
            inner.batch_depth -= 1;
            inner.batch_depth == 0
        };

        if should_flush {
            self.flush_batched_invalidations();
        }
    }

    fn is_batching(&self) -> bool {
        self.inner.borrow().batch_depth > 0
    }

    fn flush_batched_invalidations(&self) {
        // #lzspecrevisionengine: in revision mode, bump the global revision
        // once for the entire batch (O(1)), then scan effects for staleness.
        // The push-mode DFS (below) is skipped entirely.
        if self.inner.borrow().revision_mode {
            self.inner.borrow_mut().revision += 1;
            self.flush_effects_revision();
            self.inner.borrow_mut().batched_cells.clear();
            self.inner.borrow_mut().batched_cell_clears.clear();
            self.inner.borrow_mut().batched_slots.clear();
            return;
        }
        // Batch ALL invalidation/clear roots from all changed cells/slots into
        // ONE DFS pass under a SINGLE `borrow_mut` — avoids N separate
        // `borrow_mut` + DFS-queue allocations for N batched cells (#lzbatchborrow).
        let all_effects = {
            let mut inner = self.inner.borrow_mut();
            inner.batched_cells.sort_unstable();
            inner.batched_cells.dedup();
            inner.batched_cell_clears.sort_unstable();
            inner.batched_cell_clears.dedup();
            inner.batched_slots.sort_unstable();
            inner.batched_slots.dedup();

            let cells = std::mem::take(&mut inner.batched_cells);
            let cell_clears = std::mem::take(&mut inner.batched_cell_clears);
            let slots = std::mem::take(&mut inner.batched_slots);

            // Reusable effects sink: cleared once here, appended across all
            // frontier walks, then taken out (so it can outlive the borrow while
            // effects are scheduled) and restored afterward so its capacity
            // survives to the next invalidation instead of reallocating.
            inner.effects_scratch.clear();

            // Collect invalidation roots from all changed cells.
            let mut roots: Vec<SlotId> = Vec::new();
            for cell_id in &cells {
                if let Some(Node::Source(c)) = Self::get_node(&inner.nodes, *cell_id) {
                    roots.extend_from_slice(&c.dependents);
                }
            }
            Self::mark_frontier_locked(&mut inner, &roots);

            // Collect clear roots from all cleared cells.
            let mut clear_roots: Vec<SlotId> = Vec::new();
            for cell_id in &cell_clears {
                if let Some(Node::Source(c)) = Self::get_node(&inner.nodes, *cell_id) {
                    clear_roots.extend_from_slice(&c.dependents);
                }
            }
            Self::clear_frontier_locked(&mut inner, &clear_roots);

            // Clear slots directly.
            Self::clear_frontier_locked(&mut inner, &slots);

            std::mem::take(&mut inner.effects_scratch)
        };
        for (effect_id, force) in &all_effects {
            self.schedule_effect(*effect_id, *force);
        }
        self.inner.borrow_mut().effects_scratch = all_effects;
        self.flush_effects();
    }

    // -- Effect API --------------------------------------------------------

    /// Create an effect, run it immediately, and automatically rerun it after
    /// any cells/slots it read are invalidated.
    ///
    /// The callback may return `()` for no cleanup or a `FnOnce() + 'static`
    /// cleanup closure. Cleanup runs before each rerun and when the effect is
    /// disposed.
    pub fn effect<F, R>(&self, run: F) -> Effect
    where
        F: Fn(&Compute) -> R + 'static,
        R: EffectCallbackResult + 'static,
    {
        let id = self.alloc_id();
        let node = EffectNode {
            run: Rc::new(move |ctx| run(ctx).into_cleanup()),
            dependencies: EdgeVec::new(),
            cleanup: None,
            force_run: true,
        };
        self.insert_node(id, Node::Effect(node));
        self.inner.borrow_mut().all_effect_ids.push(id);
        let handle = Effect::new(id);
        self.schedule_effect(id, false);
        self.flush_effects();
        handle
    }

    /// Dispose an effect by handle.
    pub fn dispose_effect(&self, handle: &Effect) {
        let torn_down = {
            let mut inner = self.inner.borrow_mut();
            // #lzspecedgeindex: deschedule in O(1) and leave any queue entry as
            // a tombstone rather than scanning `pending_effects` for it. A mass
            // teardown during a flush disposes W effects while the queue still
            // holds W of them, so the scan was O(W) per disposal — 11491
            // ns/effect at width 65536 against 34.2 at width 16, on identical
            // total work. `pop_scheduled_effect` discards the tombstone, which
            // is what keeps a recycled id from triggering a spurious run.
            #[cfg(audit_probe)]
            crate::context::audit_probe::record_dispose_queue_len(inner.pending_effects.len());
            // Audit control: `--cfg naive_dispose_scan` restores the eager
            // O(queue) scan, so the tombstone win can be measured as a delta.
            #[cfg(naive_dispose_scan)]
            inner.pending_effects.retain(|queued| *queued != handle.id);
            Self::deschedule_effect(&mut inner, handle.id);
            let Some(Node::Effect(effect)) = Self::take_node(&mut inner.nodes, handle.id) else {
                return;
            };
            Self::remove_dependent_edges_locked(&mut inner, handle.id, &effect.dependencies);
            // The id is about to be recycled, so drop both index entries — a
            // stale one would alias the next node allocated with this id.
            Self::drop_edge_index_entries(&mut inner, handle.id);
            inner.free_ids.push(handle.id.0);
            effect
        };
        // Outside the borrow: the effect owns its run closure and everything it
        // captured, whose Drop may re-enter the context.
        let mut torn_down = torn_down;
        let cleanup = torn_down.cleanup.take();
        drop(torn_down);

        if let Some(cleanup) = cleanup {
            cleanup();
        }
    }

    /// Tear down a derived slot: detach both edge directions, clear the node,
    /// and recycle its id.
    ///
    /// Without this a slot is permanent. `Computed` is `Copy` — an id, not an
    /// owner — so dropping every handle reclaims nothing, and the node and its
    /// edge on each dependency survive for the life of the context. Under
    /// subscribe/unsubscribe churn that is unbounded growth in both memory and
    /// propagation cost: the dependent list keeps lengthening even though the
    /// live subscriber count does not.
    ///
    /// Callers must ensure nothing still reads the slot in a live compute.
    /// Reading a disposed node throws on the next recompute — the same contract
    /// as [`Context::dispose_effect`] and the JS binding's `disposeSlot`.
    pub fn dispose_slot<T>(&self, handle: &Computed<T>) {
        // `#lzcellkernel` — disposing an eager formula tears down its puller
        // first, so no poisoned effect is stranded and the `eager_by` side
        // table never aliases a recycled id.
        self.make_lazy(handle.id);
        let torn_down = {
            let mut inner = self.inner.borrow_mut();
            // Check the kind BEFORE taking: a stale handle whose id has been
            // recycled must not tear down whatever now owns it.
            if !matches!(
                Self::get_node(&inner.nodes, handle.id),
                Some(Node::Computed(_))
            ) {
                return;
            }
            let Some(Node::Computed(slot)) = Self::take_node(&mut inner.nodes, handle.id) else {
                return;
            };
            Self::remove_dependent_edges_locked(&mut inner, handle.id, &slot.dependencies);
            Self::remove_dependency_edges_locked(&mut inner, handle.id, &slot.dependents);
            Self::invalidate_disposed_dependents_locked(&mut inner, &slot.dependents);
            Self::drop_edge_index_entries(&mut inner, handle.id);
            inner.free_ids.push(handle.id.0);
            slot
        };
        // Drop the node outside the borrow. It owns the compute closure and
        // everything that closure captured, so dropping it under the borrow
        // panics if any capture's Drop re-enters the context — which a
        // self-disposing handle type does by construction.
        drop(torn_down);
    }

    /// Tear down a source cell: detach its dependents, clear the node, and
    /// recycle its id.
    ///
    /// Cells are pure sources with no dependencies, so only downstream edges
    /// need detaching. Same contract as [`Context::dispose_slot`].
    pub fn dispose_cell<T>(&self, handle: &Source<T>) {
        let mut inner = self.inner.borrow_mut();
        if !matches!(
            Self::get_node(&inner.nodes, handle.id),
            Some(Node::Source(_))
        ) {
            return;
        }
        let Some(Node::Source(cell)) = Self::take_node(&mut inner.nodes, handle.id) else {
            return;
        };
        Self::remove_dependency_edges_locked(&mut inner, handle.id, &cell.dependents);
        Self::invalidate_disposed_dependents_locked(&mut inner, &cell.dependents);
        Self::drop_edge_index_entries(&mut inner, handle.id);
        inner.free_ids.push(handle.id.0);
        drop(inner);
        // Same reason as dispose_slot: the cell owns its value, whose Drop may
        // re-enter.
        drop(cell);
    }

    /// Open a teardown scope: nodes created through it are disposed when it
    /// drops.
    ///
    /// ```
    /// # use lazily::Context;
    /// let ctx = Context::new();
    /// let topic = ctx.cell(0u64);
    /// {
    ///     let conn = ctx.scope();
    ///     let a = conn.computed(move |c| c.get(&topic) + 1);
    ///     let _b = conn.computed(move |c| c.get(&a) * 2);   // `a` captured, still Copy
    ///     assert_eq!(ctx.get(&a), 1);
    /// } // both slots disposed here
    /// ```
    ///
    /// Grouping bounds *teardown*, not visibility: a child's nodes read
    /// parent-owned or sibling-owned nodes freely.
    ///
    /// Handles stay `Copy`. The child itself is never captured by a compute
    /// closure — only handles are — so it avoids the `'static` requirement that
    /// makes a per-node borrowing wrapper impossible.
    ///
    /// Same caveat as [`Context::dispose_slot`]: dropping a scope tears down its
    /// nodes even if something outside the scope still reads them, which throws
    /// on that reader's next recompute.
    pub fn scope(&self) -> TeardownScope<'_> {
        TeardownScope {
            ctx: self,
            owned: RefCell::new(Vec::new()),
        }
    }

    /// Tear down whatever node `id` names, dispatching on its own kind.
    ///
    /// The kind is read from the arena rather than remembered by the caller, so
    /// a teardown scope stores 8 bytes per node and no tag.
    fn dispose_id(&self, id: SlotId) {
        let kind = match Self::get_node(&self.inner.borrow().nodes, id) {
            Some(Node::Computed(_)) => 0u8,
            Some(Node::Source(_)) => 1,
            Some(Node::Effect(_)) => 2,
            None => return,
        };
        match kind {
            0 => self.dispose_slot(&Computed::<()>::from_id(id)),
            1 => self.dispose_cell(&Source::<()>::from_id(id)),
            _ => self.dispose_effect(&Effect::new(id)),
        }
    }

    /// How many nodes currently depend on `node` — the size of its reverse edge
    /// set (`#lzspecedgeindex`).
    ///
    /// This is the observable the disposal contract is written against: a
    /// subscribe/unsubscribe cycle that disposes what it creates must leave this
    /// at its starting value, no matter how many cycles run. A binding that
    /// leaks shows total-ever-created here instead of live-subscriber count.
    ///
    /// Read-only and allocation-free. Returns 0 for a disposed or unknown node,
    /// and for kinds that cannot have dependents.
    ///
    /// The count is deliberately the only thing exposed — not the edge list, not
    /// the arena. Callers can assert on graph shape without being able to reach
    /// in and mutate it, and no storage strategy (linear scan, promoted index,
    /// arena layout) is pinned by the contract.
    pub fn dependent_count(&self, node: &impl GraphNode) -> usize {
        let inner = self.inner.borrow();
        match Self::get_node(&inner.nodes, node.node_id()) {
            Some(Node::Computed(slot)) => slot.dependents.len(),
            Some(Node::Source(cell)) => cell.dependents.len(),
            // Effects are pure sinks: nothing can read one.
            Some(Node::Effect(_)) | None => 0,
        }
    }

    /// How many nodes `node` currently depends on — the size of its forward edge
    /// set (`#lzspecedgeindex`).
    ///
    /// Counterpart to [`Context::dependent_count`]; disposal must detach both
    /// directions, and a binding that detaches only one leaves a dangling
    /// half-edge visible here.
    ///
    /// Returns 0 for a disposed or unknown node, and for source cells, which
    /// have no dependencies by construction.
    pub fn dependency_count(&self, node: &impl GraphNode) -> usize {
        let inner = self.inner.borrow();
        match Self::get_node(&inner.nodes, node.node_id()) {
            Some(Node::Computed(slot)) => slot.dependencies.len(),
            Some(Node::Effect(effect)) => effect.dependencies.len(),
            // Cells are pure sources.
            Some(Node::Source(_)) | None => 0,
        }
    }

    /// Check whether an effect is still registered.
    pub fn is_effect_active(&self, handle: &Effect) -> bool {
        let inner = self.inner.borrow();
        matches!(
            Self::get_node(&inner.nodes, handle.id),
            Some(Node::Effect(_))
        )
    }

    fn schedule_effect(&self, id: SlotId, force: bool) {
        let mut inner = self.inner.borrow_mut();
        let exists = match Self::get_node_mut(&mut inner.nodes, id) {
            Some(Node::Effect(effect)) => {
                if force {
                    effect.force_run = true;
                }
                true
            }
            _ => false,
        };
        if !exists {
            return;
        }

        let idx = Self::node_index(id).expect("SlotId does not fit usize");
        let already_scheduled = idx < inner.scheduled_effects.len() && inner.scheduled_effects[idx];
        if !already_scheduled {
            if idx >= inner.scheduled_effects.len() {
                inner.scheduled_effects.resize(idx + 1, false);
            }
            inner.scheduled_effects[idx] = true;
            inner.pending_effects.push_back(id);
            #[cfg(feature = "instrumentation")]
            {
                let depth = inner.pending_effects.len();
                inner.instrumentation.record_effect_queue_push(depth);
            }
        }
    }

    fn deschedule_effect(inner: &mut ContextInner, id: SlotId) {
        let idx = Self::node_index(id).expect("SlotId does not fit usize");
        if idx < inner.scheduled_effects.len() {
            inner.scheduled_effects[idx] = false;
        }
    }

    #[cfg(test)]
    fn is_effect_scheduled(&self, id: SlotId) -> bool {
        let inner = self.inner.borrow();
        let idx = Self::node_index(id).expect("SlotId does not fit usize");
        idx < inner.scheduled_effects.len() && inner.scheduled_effects[idx]
    }

    /// Drop `id` from the pending queue, if it is actually queued.
    ///
    /// #lzspecedgeindex: the scheduled-effects bitset is authoritative — an id
    /// is queued only if `schedule_effect` pushed it, and that push is gated on
    /// (and sets) the flag. So a clear flag proves absence from the queue, and
    /// the O(queue) `retain` can be skipped outright.
    ///
    /// This is the hot path: `flush_effects` pops an id and deschedules it
    /// before calling `run_effect`, so by the time the removal runs the id is
    /// provably not in the queue. Scanning for it anyway cost O(W) per effect,
    /// i.e. O(W^2) per publish — measured at 11276 ns/effect at width 65536
    /// against 27.8 at width 16, on identical total work.
    fn remove_pending_effect(&self, id: SlotId) {
        let mut inner = self.inner.borrow_mut();
        let idx = Self::node_index(id).expect("SlotId does not fit usize");
        if idx < inner.scheduled_effects.len() && inner.scheduled_effects[idx] {
            inner.pending_effects.retain(|queued| *queued != id);
            inner.scheduled_effects[idx] = false;
        }
    }

    /// Pop the next effect that is still actually scheduled, discarding
    /// tombstones.
    ///
    /// #lzspecedgeindex: `dispose_effect` cannot afford to scan the queue for
    /// the id it is removing — during a mass teardown the queue holds every
    /// sibling, so that scan is O(W) per disposal and O(W^2) overall. Instead
    /// disposal clears the scheduled flag in O(1) and leaves the queue entry
    /// behind as a tombstone, and this claims entries lazily: an entry whose
    /// flag is clear was disposed (or already run) since it was pushed, so it
    /// is dropped without running.
    ///
    /// This is what makes a stale entry safe against id recycling, which is the
    /// hazard the old eager scan existed to prevent. If the id is recycled and
    /// the new node scheduled, `schedule_effect` sets the flag and pushes
    /// again; the tombstone then claims the flag and runs the new effect once,
    /// and the second entry finds the flag clear and is discarded. Exactly one
    /// run either way.
    fn pop_scheduled_effect(inner: &mut ContextInner) -> Option<SlotId> {
        while let Some(id) = inner.pending_effects.pop_front() {
            let idx = Self::node_index(id).expect("SlotId does not fit usize");
            if idx < inner.scheduled_effects.len() && inner.scheduled_effects[idx] {
                inner.scheduled_effects[idx] = false;
                return Some(id);
            }
        }
        None
    }

    /// Drain the scheduled-effect worklist, bounded by `drain_budget`.
    ///
    /// The bound is on **iterations, not re-entry depth**. A nested
    /// `flush_effects` returns immediately at the guard below, so depth is
    /// pinned at 1 by construction and a depth bound would never fire; a
    /// scheduler-closed feedback loop is a flat unbounded drain. See
    /// `DEFAULT_DRAIN_BUDGET`.
    ///
    /// On exhaustion the drain stops and records a `DrainExhaustion` rather
    /// than panicking or spinning. Exhaustion is not convergence — the
    /// worklist is left as-is, and the report names the effects that were
    /// cycling.
    pub(crate) fn flush_effects(&self) {
        {
            let mut inner = self.inner.borrow_mut();
            if inner.flushing_effects {
                return;
            }
            inner.flushing_effects = true;
            // Only the outer drain owns the attribution buffer; clearing here
            // (not per iteration) keeps a terminating flush free of scans.
            inner.drain_runs.clear();
        }

        let mut iterations: usize = 0;
        loop {
            let id = {
                let mut inner = self.inner.borrow_mut();
                match Self::pop_scheduled_effect(&mut inner) {
                    Some(id) => id,
                    None => {
                        inner.flushing_effects = false;
                        return;
                    }
                }
            };

            {
                let mut inner = self.inner.borrow_mut();
                if let Some(idx) = Self::node_index(id) {
                    if idx >= inner.drain_runs.len() {
                        inner.drain_runs.resize(idx + 1, 0);
                    }
                    inner.drain_runs[idx] = inner.drain_runs[idx].saturating_add(1);
                }

                iterations += 1;
                if iterations >= inner.drain_budget {
                    let report = Self::drain_exhaustion_report(&inner, iterations);
                    inner.last_drain_exhaustion = Some(report);
                    inner.flushing_effects = false;
                    return;
                }
            }

            self.run_effect(id);
        }
    }

    /// Build the exhaustion report: the busiest effects in the current drain,
    /// descending. Capped at a handful of entries — the head is what
    /// identifies a scheduler-closed loop, and a full histogram over a wide
    /// graph would bury it.
    fn drain_exhaustion_report(inner: &ContextInner, iterations: usize) -> DrainExhaustion {
        const TOP_N: usize = 8;
        let mut top: Vec<(u64, u32)> = inner
            .drain_runs
            .iter()
            .enumerate()
            .filter(|&(_, &runs)| runs > 0)
            .map(|(idx, &runs)| (idx as u64, runs))
            .collect();
        top.sort_unstable_by(|a, b| b.1.cmp(&a.1).then(a.0.cmp(&b.0)));
        top.truncate(TOP_N);
        DrainExhaustion {
            iterations,
            budget: inner.drain_budget,
            top_effects: top,
        }
    }

    /// The most recent drain exhaustion, if any drain has been cut short.
    ///
    /// `None` means every drain so far ended with an empty worklist. This is
    /// the observable a conformance fixture asserts on: a divergent
    /// scheduler-closed loop must surface here rather than hang.
    pub fn last_drain_exhaustion(&self) -> Option<DrainExhaustion> {
        self.inner.borrow().last_drain_exhaustion.clone()
    }

    /// Clear the recorded exhaustion, so a later drain can be observed
    /// independently.
    pub fn clear_drain_exhaustion(&self) {
        self.inner.borrow_mut().last_drain_exhaustion = None;
    }

    /// Current effect-drain iteration budget.
    pub fn drain_budget(&self) -> usize {
        self.inner.borrow().drain_budget
    }

    /// Override the effect-drain iteration budget.
    ///
    /// Lowering it is how a test exercises divergence without waiting for
    /// `DEFAULT_DRAIN_BUDGET` iterations. Raising it is a blunt instrument: if
    /// a legitimate cascade trips the default, prefer reading
    /// `DrainExhaustion::top_effects` to confirm the runs are spread across
    /// effects rather than concentrated in one, because a concentrated head is
    /// divergence and a larger budget only delays it.
    ///
    /// # Panics
    /// If `budget` is zero — a zero budget would abort every drain before it
    /// ran a single effect.
    pub fn set_drain_budget(&self, budget: usize) {
        assert!(budget > 0, "drain budget must be non-zero");
        self.inner.borrow_mut().drain_budget = budget;
    }

    /// #lzspecrevisionengine: revision-mode effect flush. Scans all registered
    /// effects and schedules those whose dependencies are stale
    /// (`verified_at < revision`). O(effects) per flush, not O(cone) per write —
    /// the effect-side cost is decoupled from the write-path cone walk that
    /// revision mode eliminates.
    fn flush_effects_revision(&self) {
        let stale_effects: Vec<SlotId> = {
            let inner = self.inner.borrow();
            inner
                .all_effect_ids
                .iter()
                .filter(|&&eid| match Self::get_node(&inner.nodes, eid) {
                    Some(Node::Effect(e)) => {
                        e.dependencies.iter().any(|dep| {
                            matches!(Self::get_node(&inner.nodes, *dep),
                                Some(Node::Computed(s)) if s.verified_at < inner.revision)
                        }) || e.force_run
                    }
                    _ => false,
                })
                .copied()
                .collect()
        };
        for eid in stale_effects {
            self.schedule_effect(eid, true);
        }
        self.flush_effects();
    }

    fn run_effect(&self, id: SlotId) {
        if !self.effect_should_run(id) {
            return;
        }
        self.remove_pending_effect(id);

        let run: Rc<EffectFn>;
        let old_deps;
        let cleanup: Option<Box<dyn FnOnce()>>;
        {
            let mut inner = self.inner.borrow_mut();
            let effect = match Self::get_node_mut(&mut inner.nodes, id) {
                Some(Node::Effect(effect)) => effect,
                _ => return,
            };
            old_deps = std::mem::take(&mut effect.dependencies);
            cleanup = effect.cleanup.take();
            effect.force_run = false;
            run = Rc::clone(&effect.run);
            if !inner.dependencies_index.is_empty() {
                inner.dependencies_index.remove(&id);
            }
            Self::remove_dependent_edges_locked(&mut inner, id, &old_deps);
        }

        if let Some(cleanup) = cleanup {
            cleanup();
        }

        let next_cleanup = {
            let cx = Compute::new(self, id, 0);
            (run.as_ref())(&cx)
        };

        let mut inner = self.inner.borrow_mut();
        if let Some(Node::Effect(effect)) = Self::get_node_mut(&mut inner.nodes, id) {
            effect.cleanup = next_cleanup;
        } else if let Some(cleanup) = next_cleanup {
            drop(inner);
            cleanup();
        }
    }

    fn effect_should_run(&self, id: SlotId) -> bool {
        let (force_run, dependencies) = {
            let inner = self.inner.borrow();
            let Some(Node::Effect(effect)) = Self::get_node(&inner.nodes, id) else {
                return false;
            };
            (effect.force_run, effect.dependencies.clone())
        };

        if force_run {
            return true;
        }

        dependencies
            .into_iter()
            .any(|dep_id| self.is_slot_node(dep_id) && self.refresh_slot(dep_id))
    }

    // -- Signal API --------------------------------------------------------

    /// Create an **eager** derived value that recomputes immediately whenever
    /// one of its dependencies is invalidated.
    ///
    /// Where [`Context::computed`] is lazy (recomputed on the next read), a
    /// signal is materialized eagerly: by the time the invalidating
    /// `set_cell`/`set`/`batch` call returns, the signal already holds its new
    /// value. The value is always set, so observers never see an intermediate
    /// unset state — a dependency change drives the value directly from `v1`
    /// to `v2`.
    ///
    /// The signal is backed by a memoized slot, so a recomputation that yields
    /// an equal value (via `PartialEq`) does not invalidate downstream
    /// dependents. Recomputation is pull-based and therefore glitch-free: a
    /// signal that reads other signals/slots always observes values consistent
    /// with the current inputs.
    pub fn signal<T, F>(&self, compute: F) -> Computed<T>
    where
        T: PartialEq + 'static,
        F: Fn(&Compute) -> T + 'static,
    {
        // `#lzcellkernel` — the eager construction is a guarded computed cell
        // that is then made eager: `computed(f).eager()`. Retires the former
        // two-node `Signal` (guarded slot + puller effect); the coalescing comes
        // from the scheduler, so a per-write puller cannot be built.
        let computed = self.computed(compute);
        self.make_eager::<T>(computed.id);
        computed
    }

    /// Read an eager formula's current value. Always returns a materialized
    /// value. (Was `get_signal`; an eager formula reads with ordinary `get`.)
    pub fn get_signal<T: Clone + 'static>(&self, handle: &Computed<T>) -> T {
        self.get(handle)
    }

    /// Read an eager formula's current value as `Rc<T>`, avoiding a deep clone.
    pub fn get_signal_rc<T: 'static>(&self, handle: &Computed<T>) -> Rc<T> {
        self.get_rc(handle)
    }

    /// Make a formula lazy (revert from eager). Stops eager recomputation; the
    /// backing value remains readable and recomputes on next read. (Was
    /// `dispose_signal`; now the `lazy` reverse transition.)
    pub fn dispose_signal<T>(&self, handle: &Computed<T>) {
        self.make_lazy(handle.id);
    }

    /// Check whether a formula is currently eager.
    pub fn is_signal_active<T>(&self, handle: &Computed<T>) -> bool {
        self.is_eager(handle.id)
    }

    // -- Clearing ----------------------------------------------------------

    /// Hard-clear a slot's cached value and recursively clear all dependents.
    pub(crate) fn clear_slot(&self, id: SlotId) {
        if self.is_batching() {
            self.inner.borrow_mut().batched_slots.push(id);
            return;
        }
        self.clear_slot_now(id);
    }

    /// Batch-aware multi-root slot invalidation. Clears each id's cached value
    /// and recursively clears dependents in ONE frontier walk, then flushes any
    /// scheduled effects exactly once. Used by demand-driven derived readers
    /// (e.g. [`QueueCell`](crate::QueueCell) reader-kinds) that own an
    /// out-of-graph mutation source and must invalidate several derived slots
    /// atomically on a single op — a push/pop whose `len`/`is_full` transition
    /// together must never glitch. Unsubscribed + uncached roots hit the
    /// `clear_frontier` no-op fast path, so an op nobody observes costs ~O(roots)
    /// with no derivation, no effect scheduling, and no flush.
    pub(crate) fn clear_slots(&self, ids: &[SlotId]) {
        if ids.is_empty() {
            return;
        }
        if self.is_batching() {
            self.inner.borrow_mut().batched_slots.extend_from_slice(ids);
            return;
        }
        let effects_to_schedule = {
            let mut inner = self.inner.borrow_mut();
            inner.effects_scratch.clear();
            Self::clear_frontier_locked(&mut inner, ids);
            std::mem::take(&mut inner.effects_scratch)
        };
        // Store-without-cascade (read-side dual): if clearing the roots reached
        // no Effect, there is nothing to flush — an unobserved op skips the
        // flush machinery entirely and returns after a single frontier walk.
        if !effects_to_schedule.is_empty() {
            for (effect_id, force) in effects_to_schedule.iter().copied() {
                self.schedule_effect(effect_id, force);
            }
            self.flush_effects();
        }
        // Restore the reusable sink so its capacity survives to the next call.
        self.inner.borrow_mut().effects_scratch = effects_to_schedule;
    }

    pub(crate) fn flush_effects_after_invalidation(&self) {
        if !self.is_batching() {
            self.flush_effects();
        }
    }

    fn clear_slot_now(&self, id: SlotId) {
        let effects_to_schedule = {
            let mut inner = self.inner.borrow_mut();
            let roots = [id];
            inner.effects_scratch.clear();
            Self::clear_frontier_locked(&mut inner, &roots);
            std::mem::take(&mut inner.effects_scratch)
        };
        for (effect_id, force) in effects_to_schedule.iter().copied() {
            self.schedule_effect(effect_id, force);
        }
        self.inner.borrow_mut().effects_scratch = effects_to_schedule;
    }

    pub(crate) fn clear_cell_dependents(&self, id: SlotId) {
        if self.is_batching() {
            self.inner.borrow_mut().batched_cell_clears.push(id);
            return;
        }
        self.clear_cell_dependents_now(id);
        self.flush_effects();
    }

    /// Returns `true` iff at least one Effect was scheduled (i.e. the dependent
    /// cone contains an active reactor that must flush). A `false` result is the
    /// store-without-cascade fast path: the value is already stored and lazy Slot
    /// dependents are dirty-marked, but no effect flush is owed.
    fn invalidate_cell_dependents_now(&self, id: SlotId) -> bool {
        self.invalidate_dependents_now(id)
    }

    fn clear_cell_dependents_now(&self, id: SlotId) {
        let effects_to_schedule = {
            let mut inner = self.inner.borrow_mut();
            let roots = match Self::get_node(&inner.nodes, id) {
                Some(Node::Source(c)) => c.dependents.clone(),
                _ => return,
            };
            inner.effects_scratch.clear();
            Self::clear_frontier_locked(&mut inner, &roots);
            std::mem::take(&mut inner.effects_scratch)
        };
        for (effect_id, force) in effects_to_schedule.iter().copied() {
            self.schedule_effect(effect_id, force);
        }
        self.inner.borrow_mut().effects_scratch = effects_to_schedule;
    }

    /// Batched BFS invalidation: marks all reachable slots dirty under a SINGLE
    /// `borrow_mut`, then schedules collected effects after the borrow is released.
    /// Replaces the former recursive `mark_slot_dirty` / `invalidate_dependent_from_changed_value`
    /// which re-borrowed per node — for fan-out 256 this cuts ~768 RefCell operations
    /// to 1 (#lzbatchborrow).
    fn invalidate_dependents_now(&self, id: SlotId) -> bool {
        let effects_to_schedule = {
            let mut inner = self.inner.borrow_mut();
            let inner_mut = &mut *inner;
            // Swap the dependent list into a reused buffer instead of cloning
            // it. Cloning cost 8 bytes per dependent plus an allocation on
            // every set_cell — 10.7ms per publish at width 1M — even when
            // nothing downstream was read. The graph is acyclic, so marking
            // cannot reach `id` again and observe the borrowed-out list.
            match Self::get_node_mut(&mut inner_mut.nodes, id) {
                Some(Node::Source(c)) if c.dependents.is_empty() => return false,
                Some(Node::Computed(s)) if s.dependents.is_empty() => return false,
                Some(Node::Source(c)) => {
                    std::mem::swap(&mut c.dependents, &mut inner_mut.roots_scratch)
                }
                Some(Node::Computed(s)) => {
                    std::mem::swap(&mut s.dependents, &mut inner_mut.roots_scratch)
                }
                _ => return false,
            }
            let roots = std::mem::take(&mut inner.roots_scratch);
            inner.effects_scratch.clear();
            Self::mark_frontier_locked(&mut inner, &roots);
            inner.roots_scratch = roots;
            // put the list back on its node
            let inner_mut = &mut *inner;
            match Self::get_node_mut(&mut inner_mut.nodes, id) {
                Some(Node::Source(c)) => {
                    std::mem::swap(&mut c.dependents, &mut inner_mut.roots_scratch)
                }
                Some(Node::Computed(s)) => {
                    std::mem::swap(&mut s.dependents, &mut inner_mut.roots_scratch)
                }
                _ => {}
            }
            std::mem::take(&mut inner.effects_scratch)
        };
        let scheduled = !effects_to_schedule.is_empty();
        for (effect_id, force) in effects_to_schedule.iter().copied() {
            self.schedule_effect(effect_id, force);
        }
        self.inner.borrow_mut().effects_scratch = effects_to_schedule;
        scheduled
    }

    /// Single-borrow DFS dirty-marking. Roots get `force=true`; transitive
    /// descendants get `force=false` (matching the former recursive semantics).
    /// Appends `(effect_id, force)` pairs to `effects_scratch` for the caller to
    /// schedule after the borrow is released. Reuses the `ContextInner`'s
    /// `mark_scratch`/`effects_scratch` buffers so an invalidation no longer
    /// allocates a DFS stack + force stack per call: the former separate
    /// `stack` and `force_stack` collapse into one `Vec<(SlotId, bool)>` with
    /// better pop locality (#lzbatchborrow).
    /// Dirty the cone that read a node being disposed (`#lzspecedgeindex`).
    ///
    /// Detaching the edges is not enough on its own: a dependent that already
    /// has a cached value would keep serving it forever, since with its
    /// dependency edge gone nothing will ever invalidate it again — not even a
    /// later publish on the disposed node's own source. The spec requires that
    /// reader to error on its next recompute, so the cone must be marked dirty
    /// and recompute rather than answer from cache.
    ///
    /// Effects reached by the walk are deliberately NOT scheduled. Disposal is
    /// not a publish: an effect's next recompute is driven by a real write, and
    /// running one here would re-enter a compute that reads the node currently
    /// being torn down, turning `dispose` itself into a panic and breaking
    /// teardown idempotence.
    fn invalidate_disposed_dependents_locked(inner: &mut ContextInner, dependents: &[SlotId]) {
        if dependents.is_empty() {
            return;
        }
        inner.effects_scratch.clear();
        Self::mark_frontier_locked(inner, dependents);
        inner.effects_scratch.clear();
    }

    fn mark_frontier_locked(inner: &mut ContextInner, roots: &[SlotId]) {
        let nodes = &mut inner.nodes;
        let stack = &mut inner.mark_scratch;
        let effects = &mut inner.effects_scratch;
        stack.clear();
        for &root in roots {
            stack.push((root, true));
        }
        while let Some((id, force)) = stack.pop() {
            match Self::get_node_mut(nodes, id) {
                Some(Node::Computed(slot)) => {
                    let should_propagate = !slot.dirty || (force && !slot.force_recompute);
                    slot.dirty = true;
                    if force {
                        slot.force_recompute = true;
                    }
                    if should_propagate {
                        for dep_id in &slot.dependents {
                            stack.push((*dep_id, false));
                        }
                    }
                }
                Some(Node::Effect(_)) => {
                    effects.push((id, force));
                }
                _ => {}
            }
        }
    }

    /// Single-borrow DFS value-clearing. Clears slot values and dirty flags
    /// recursively, appending effects to schedule to `effects_scratch`.
    fn clear_frontier_locked(inner: &mut ContextInner, roots: &[SlotId]) {
        let nodes = &mut inner.nodes;
        let stack = &mut inner.mark_scratch;
        let effects = &mut inner.effects_scratch;
        stack.clear();
        for &root in roots {
            stack.push((root, true));
        }
        while let Some((id, _)) = stack.pop() {
            match Self::get_node_mut(nodes, id) {
                Some(Node::Computed(slot)) => {
                    if slot.value.is_none() && !slot.dirty {
                        continue;
                    }
                    slot.value = AnyValue::None;
                    slot.dirty = false;
                    slot.force_recompute = false;
                    for dep_id in &slot.dependents {
                        stack.push((*dep_id, true));
                    }
                }
                Some(Node::Effect(_)) => {
                    effects.push((id, true));
                }
                _ => {}
            }
        }
    }

    fn notify_slot_value_changed(&self, id: SlotId) {
        // #lzspecrevisionengine: in revision mode, downstream slots detect
        // staleness via `verified_at < revision` (the global revision was bumped
        // on the write). No dirty walk needed — the cone walk is the push cost
        // revision mode eliminates.
        if self.inner.borrow().revision_mode {
            return;
        }
        self.invalidate_dependents_now(id);
    }

    /// Check whether a slot currently has a cached, fresh value (for testing).
    pub fn is_set<T: 'static>(&self, handle: &Computed<T>) -> bool {
        let inner = self.inner.borrow();
        if let Some(Node::Computed(slot)) = Self::get_node(&inner.nodes, handle.id) {
            !slot.value.is_none() && !slot.dirty
        } else {
            false
        }
    }

    fn factory_handle<H>(&self, key: FactoryKey, value_type: TypeId) -> Option<H>
    where
        H: Copy + 'static,
    {
        let inner = self.inner.borrow();
        let entry = inner.factory_handles.get(&key)?;
        assert!(
            entry.value_type == value_type,
            "lazily: factory key {:?} was reused with an incompatible value type",
            key.factory_type
        );
        Some(
            *entry
                .handle
                .downcast_ref::<H>()
                .expect("lazily: factory handle type mismatch"),
        )
    }

    fn insert_factory_handle<H>(&self, key: FactoryKey, value_type: TypeId, handle: H)
    where
        H: Copy + 'static,
    {
        self.inner.borrow_mut().factory_handles.insert(
            key,
            FactoryEntry {
                value_type,
                handle: Rc::new(handle),
            },
        );
    }

    /// Return the current benchmark instrumentation counters.
    #[cfg(feature = "instrumentation")]
    pub fn instrumentation_snapshot(&self) -> crate::instrumentation::InstrumentationSnapshot {
        self.inner.borrow().instrumentation.snapshot()
    }

    /// Reset benchmark instrumentation counters to zero.
    #[cfg(feature = "instrumentation")]
    pub fn reset_instrumentation(&self) {
        self.inner.borrow_mut().instrumentation.reset();
    }
}

impl Default for Context {
    fn default() -> Self {
        Self::new()
    }
}

pub(crate) mod sealed {
    pub trait Sealed {}
}

/// A node in a [`Context`]'s reactive graph, addressed by one of its handles.
///
/// Sealed: implemented for [`Computed`], [`Source`], and [`Effect`]
/// only. It exists so the graph-shape accessors take any handle kind without
/// exposing the internal node id, and cannot be implemented downstream.
pub trait GraphNode: sealed::Sealed {
    #[doc(hidden)]
    fn node_id(&self) -> SlotId;
}

// `#lzcellkernel` — one impl per concrete handle struct. Every cell (a `Source`
// or a `Computed`) is a graph node addressed by its slot id, regardless of kind.
impl<T, M> sealed::Sealed for Source<T, M> {}
impl<T, M> GraphNode for Source<T, M> {
    fn node_id(&self) -> SlotId {
        self.id
    }
}
impl<T> sealed::Sealed for Computed<T> {}
impl<T> GraphNode for Computed<T> {
    fn node_id(&self) -> SlotId {
        self.id
    }
}

impl sealed::Sealed for Effect {}
impl GraphNode for Effect {
    fn node_id(&self) -> SlotId {
        self.id
    }
}

/// A teardown scope over a [`Context`]: nodes created through it are disposed
/// when it drops. See [`Context::child`].
///
/// Records only ids — 8 bytes per node, no boxing — and reads each node's kind
/// from the arena at teardown.
pub struct TeardownScope<'ctx> {
    ctx: &'ctx Context,
    owned: RefCell<Vec<SlotId>>,
}

impl TeardownScope<'_> {
    /// Create a guarded computed cell owned by this scope.
    pub fn computed<T, F>(&self, compute: F) -> Computed<T>
    where
        T: PartialEq + 'static,
        F: Fn(&Compute) -> T + 'static,
    {
        let handle = self.ctx.computed(compute);
        self.owned.borrow_mut().push(handle.id);
        handle
    }

    /// Create a source cell owned by this scope.
    pub fn source<T: PartialEq + 'static>(&self, value: T) -> Source<T> {
        let handle = self.ctx.source(value);
        self.owned.borrow_mut().push(handle.id);
        handle
    }

    /// Compatibility constructor for the pre-Cell-kernel API.
    #[deprecated(note = "use `TeardownScope::source`")]
    pub fn cell<T: PartialEq + 'static>(&self, value: T) -> Source<T> {
        self.source(value)
    }

    /// Register an effect owned by this scope.
    pub fn effect<F, R>(&self, run: F) -> Effect
    where
        F: Fn(&Compute) -> R + 'static,
        R: EffectCallbackResult + 'static,
    {
        let handle = self.ctx.effect(run);
        self.owned.borrow_mut().push(handle.id);
        handle
    }

    /// The context this scope belongs to.
    pub fn context(&self) -> &Context {
        self.ctx
    }

    /// How many nodes this scope owns.
    pub fn len(&self) -> usize {
        self.owned.borrow().len()
    }

    /// Whether this scope owns nothing.
    pub fn is_empty(&self) -> bool {
        self.len() == 0
    }

    /// Disarm the scope: it is armed to dispose its nodes when it ends, and
    /// this turns that off. Ending a disarmed scope disposes nothing, and its
    /// nodes revert to context ownership — the state every unscoped node is
    /// already in.
    ///
    /// The nodes themselves are untouched. The only thing that changes is
    /// whether this scope fires at end-of-life, which is what `disarm` names —
    /// the same sense as defusing a scope guard.
    pub fn disarm(self) {
        self.owned.borrow_mut().clear();
    }
}

impl Drop for TeardownScope<'_> {
    fn drop(&mut self) {
        // Reverse creation order: dependents before what they read, so a scope
        // does not transiently dangle inside itself while tearing down.
        let owned = std::mem::take(&mut *self.owned.borrow_mut());
        for id in owned.into_iter().rev() {
            self.ctx.dispose_id(id);
        }
    }
}

// -- Capability trait impls (#lzspecedgeindex) -------------------------------
//
// These land after the inherent methods they forward to: the per-context
// disposal work predates the abstraction it now satisfies.

impl crate::reactive_graph::Teardown for TeardownScope<'_> {
    fn len(&self) -> usize {
        TeardownScope::len(self)
    }
    fn disarm(self) {
        TeardownScope::disarm(self);
    }
}

impl crate::reactive_graph::ReactiveGraph for Context {
    type Computed<T> = crate::cell::Computed<T>;
    type Source<T> = crate::cell::Source<T>;
    type Effect = crate::effect::Effect;
    type Scope<'a> = TeardownScope<'a>;

    fn dispose_slot<T: 'static>(&self, handle: &Self::Computed<T>) {
        Context::dispose_slot(self, handle);
    }
    fn dispose_cell<T: 'static>(&self, handle: &Self::Source<T>) {
        Context::dispose_cell(self, handle);
    }
    fn dispose_effect(&self, handle: &Self::Effect) {
        Context::dispose_effect(self, handle);
    }
    fn scope(&self) -> Self::Scope<'_> {
        Context::scope(self)
    }
    fn batch<R>(&self, run: impl FnOnce(&Self) -> R) -> R {
        Context::batch(self, run)
    }
    fn dependent_count(&self, node: &impl GraphNode) -> usize {
        Context::dependent_count(self, node)
    }
    fn dependency_count(&self, node: &impl GraphNode) -> usize {
        Context::dependency_count(self, node)
    }
}

impl crate::reactive_graph::SyncReactiveGraph for Context {
    // The value-threaded per-recompute view (`#lzcellkernel`): the capability
    // trait now hands its `computed`/`effect` closures a `&Compute` directly, so
    // there is no ambient thread-local frame to bridge.
    type Compute<'a> = Compute<'a>;

    fn source<T>(&self, value: T) -> Self::Source<T>
    where
        T: PartialEq + Send + Sync + 'static,
    {
        Context::source(self, value)
    }
    fn computed<T, F>(&self, compute: F) -> Self::Computed<T>
    where
        T: PartialEq + Send + Sync + 'static,
        F: Fn(&Self::Compute<'_>) -> T + Send + Sync + 'static,
    {
        // Value-threaded (`#lzcellkernel`): the closure receives the fortified
        // `&Compute` view directly, so its tracked reads attribute to the
        // recomputing node with no ambient frame. The old thread-local bridge is
        // gone.
        Context::computed(self, compute)
    }
    fn effect<F, C>(&self, run: F) -> Self::Effect
    where
        F: Fn(&Self::Compute<'_>) -> C + Send + Sync + 'static,
        C: FnOnce() + Send + Sync + 'static,
    {
        // Same value-threaded view as `computed` above.
        Context::effect(self, run)
    }
}

/// A handle readable through a context's `get` (`#lzcellkernel`), generic over
/// the context type `Ctx`.
///
/// One trait for every context (`Context`, `ThreadSafeContext`, `AsyncContext`,
/// …). `Output` carries the read result — usually the value `T`, but
/// `Option<T>` for an async computed that may still be pending. Implemented by
/// both [`Computed`] (recompute-if-needed) and [`Source`] (stored value); the
/// unified read that supersedes the deprecated `get_cell`.
pub trait Read<Ctx: ?Sized> {
    type Output;
    fn read(&self, ctx: &Ctx) -> Self::Output;
}

/// A handle writable through a context's `set` (`#lzcellkernel`), generic over
/// the context type `Ctx`.
///
/// Implemented only by source handles — writes do not exist on [`Computed`], so
/// `ctx.set(computed, v)` fails to compile (write protection, design §3).
pub trait Write<Ctx: ?Sized> {
    type Value;
    fn write(&self, ctx: &Ctx, value: Self::Value);
}

impl<T: Clone + 'static> Read<Context> for Computed<T> {
    type Output = T;
    fn read(&self, ctx: &Context) -> T {
        ctx.get_slot(self.id)
    }
}

impl<T: Clone + 'static, M> Read<Context> for Source<T, M> {
    type Output = T;
    fn read(&self, ctx: &Context) -> T {
        // A bare `&Context` read is untracked; the primary path tracks via
        // `Read<Compute>`.
        ctx.read_source_untracked(self.id)
    }
}

impl<T: PartialEq + 'static> Write<Context> for Source<T> {
    type Value = T;
    fn write(&self, ctx: &Context, value: T) {
        ctx.set_source::<T>(self.id, value);
    }
}

/// A fortified, non-escapable **compute context** (`#lzcellkernel`).
///
/// Handed to a `computed` / `slot` / `computed_ripple_when` recompute closure as
/// the **sole tracking-read surface**. It carries the recomputing node id **as a
/// value** — not an ambient thread-local — so every dependency edge registered
/// through it attributes to the *correct* node, including a read that happens
/// after an `.await` in the async bindings (the value survives suspension where
/// a thread-local would be clobbered).
///
/// Fortification, by construction:
/// - **Non-escapable:** lifetime `'a` binds it to the recompute call; it is
///   `!Send` and holds a borrow, so it cannot be stored past the closure and
///   reused to register an edge against the wrong node.
/// - **Sole surface:** the closure receives *only* a `Compute`; a normal read
///   therefore *cannot* miss tracking. An intentional untracked read is the
///   explicitly-named [`Compute::untracked`].
/// - **Generation-stamped:** `gen` pins the node's liveness at recompute start,
///   so a read against a disposed/recycled node is detected, never misattributed.
///
/// The escape hatch is closed by the type system. A `Compute` cannot be smuggled
/// out of its closure into outer state — the borrow it carries does not outlive
/// the recompute call:
///
/// ```compile_fail
/// use std::cell::RefCell;
/// use lazily::{Compute, Context};
/// let ctx = Context::new();
/// let leaked: RefCell<Option<&Compute<'_>>> = RefCell::new(None);
/// let _ = ctx.computed(|c| {
///     *leaked.borrow_mut() = Some(c); // ERROR: `c` does not live long enough
///     0i32
/// });
/// ```
///
/// Nor can it cross a thread boundary — it is `!Send`:
///
/// ```compile_fail
/// use lazily::{Compute, Context};
/// fn assert_send<T: Send>(_: &T) {}
/// let ctx = Context::new();
/// let _ = ctx.computed(|c: &Compute<'_>| {
///     assert_send(c); // ERROR: `*const ()` cannot be sent between threads safely
///     0i32
/// });
/// ```
pub struct Compute<'a> {
    ctx: &'a Context,
    slot_id: SlotId,
    #[allow(dead_code)]
    generation: u64,
    // Not `Send`/`Sync` and cannot outlive `'a` — enforces non-escapability.
    _not_send: std::marker::PhantomData<*const ()>,
}

impl<'a> Compute<'a> {
    pub(crate) fn new(ctx: &'a Context, slot_id: SlotId, generation: u64) -> Self {
        Self {
            ctx,
            slot_id,
            generation,
            _not_send: std::marker::PhantomData,
        }
    }

    /// The `SlotId` of the node currently being recomputed — the dependent every
    /// tracked read through this context subscribes.
    pub(crate) fn slot_id(&self) -> SlotId {
        self.slot_id
    }

    /// Read a cell, registering a dependency edge against this recompute's node.
    pub fn get<H: Read<Self> + ?Sized>(&self, handle: &H) -> <H as Read<Self>>::Output {
        handle.read(self)
    }

    /// Write a source cell from inside a compute (untracked — a write is an
    /// argument, never a dependency; §9.2.3).
    pub fn set<H: Write<Self> + ?Sized>(&self, handle: &H, value: <H as Write<Self>>::Value) {
        handle.write(self, value)
    }

    /// The explicit untracked-read escape: the owning [`Context`], whose reads
    /// register no dependency edge. Use only when a read deliberately must not
    /// form an edge. Returns the borrow with the compute's own lifetime `'a`.
    pub fn untracked(&self) -> &'a Context {
        self.ctx
    }

    /// Read a computed's value as `Rc<T>`, registering an edge (`#lzrsgetarc`).
    pub fn get_rc<T: 'static>(&self, handle: &Computed<T>) -> Rc<T> {
        self.ctx.register_dependency(handle.id, self.slot_id);
        self.ctx.get_rc_untracked(handle)
    }

    /// Read a source cell's value as `Rc<T>`, registering an edge against this
    /// recompute's node. The `Compute`-side of [`Context::get_cell_rc`].
    pub fn get_cell_rc<T: 'static>(&self, handle: &Source<T>) -> Rc<T> {
        self.ctx.register_dependency(handle.id, self.slot_id);
        // The `&Context` delegate is untracked, so this registers exactly one
        // edge — against this recompute's node.
        self.ctx.get_cell_rc(handle)
    }

    /// Create a source cell (construction is not a read; delegates to the ctx).
    pub fn cell<T: PartialEq + 'static>(&self, value: T) -> Source<T> {
        self.ctx.cell(value)
    }

    /// Create a source cell (v2 spelling of [`Compute::cell`]).
    pub fn source<T: PartialEq + 'static>(&self, value: T) -> Source<T> {
        self.ctx.source(value)
    }

    /// Create a guarded computed cell.
    pub fn computed<T, F>(&self, compute: F) -> Computed<T>
    where
        T: PartialEq + 'static,
        F: Fn(&Compute) -> T + 'static,
    {
        self.ctx.computed(compute)
    }

    /// Create a guarded computed with an explicit change predicate.
    pub fn computed_ripple_when<T, F, C>(&self, compute: F, changed: C) -> Computed<T>
    where
        T: 'static,
        F: Fn(&Compute) -> T + 'static,
        C: Fn(&T, &T) -> bool + 'static,
    {
        self.ctx.computed_ripple_when(compute, changed)
    }

    /// Create a pass-through (unguarded) derived cell.
    pub fn slot<T, F>(&self, compute: F) -> Computed<T>
    where
        T: 'static,
        F: Fn(&Compute) -> T + 'static,
    {
        self.ctx.slot(compute)
    }

    /// Run `f` in a batch (coalesced invalidation). The callback receives the
    /// owning [`Context`] — a batch is a write-coalescing scope.
    pub fn batch<F, R>(&self, f: F) -> R
    where
        F: FnOnce(&Context) -> R,
    {
        self.ctx.batch(f)
    }
}

impl<'a, T: Clone + 'static> Read<Compute<'a>> for Computed<T> {
    type Output = T;
    fn read(&self, cx: &Compute<'a>) -> T {
        cx.ctx.register_dependency(self.id, cx.slot_id);
        cx.ctx.get_slot_untracked(self.id)
    }
}

impl<'a, T: Clone + 'static, M> Read<Compute<'a>> for Source<T, M> {
    type Output = T;
    fn read(&self, cx: &Compute<'a>) -> T {
        cx.ctx.register_dependency(self.id, cx.slot_id);
        cx.ctx.read_source_untracked(self.id)
    }
}

impl<'a, T: PartialEq + 'static> Write<Compute<'a>> for Source<T> {
    type Value = T;
    fn write(&self, cx: &Compute<'a>, value: T) {
        cx.ctx.set_source::<T>(self.id, value);
    }
}

/// The **compute-time operations subset** of the `Context` API (`#lzcellkernel`)
/// — exactly the operations a compute/effect closure may perform (`get` / `set`
/// / `source` / `cell` / `computed` / `computed_ripple_when` / `slot` / `effect`
/// / `batch` / `get_rc` / `read_value` / `dispose_node`), and no more. The rest
/// of `Context` — revision control, degree/graph introspection, teardown-scope
/// creation, instrumentation, IPC hooks — is deliberately **off** this trait.
///
/// This is the trait that dissolves the `&Context`-vs-`&Compute` cascade: every
/// signature reachable from *inside* a compute/effect closure is written
/// against `&impl ComputeOps` / `<C: ComputeOps>` instead of a concrete
/// context, so the same handle convenience methods and internal builders work
/// whether they run at top level (through [`Context`], **untracked**) or inside
/// a recompute (through [`Compute`], **tracked** against the recomputing node).
///
/// It is deliberately distinct from the execution-model traits
/// ([`ReactiveGraph`](crate::ReactiveGraph) / `SyncReactiveGraph`): those pick
/// *which* graph implementation you hold, across models, with associated types;
/// `ComputeOps` is **intra-model** (concrete `Context` types, no associated
/// types) and names the read/construct operations the single-threaded graph
/// shares between its ambient view and its per-compute view. Read discipline is
/// the whole point — a `Compute::read_value` registers a dependency edge; a
/// `Context::read_value` does not.
///
/// **Implemented by exactly two types**, and it is not a universal funnel:
/// - [`Context`] — the owning single-threaded graph (also implements
///   [`SyncReactiveGraph`](crate::SyncReactiveGraph)); its reads are untracked.
/// - [`Compute`] — the per-recompute *view* (a view, not a model; implements
///   *only* `ComputeOps`); its reads track against the recomputing node.
pub trait ComputeOps {
    /// Read a node's value with this surface's tracking discipline: a
    /// [`Compute`] registers a dependency edge against the recomputing node; a
    /// bare [`Context`] registers nothing.
    fn read_value<T: Clone + 'static>(&self, id: SlotId) -> T;

    /// Read a computed node's value as `Rc<T>`, with the same tracking
    /// discipline as [`read_value`](Self::read_value).
    fn get_rc<T: 'static>(&self, handle: &Computed<T>) -> Rc<T>;

    /// Read a handle's value through its [`Read`] impl for this surface.
    fn get<H: Read<Self> + ?Sized>(&self, handle: &H) -> <H as Read<Self>>::Output
    where
        Self: Sized,
    {
        handle.read(self)
    }

    /// Write a source handle through its [`Write`] impl for this surface.
    fn set<H: Write<Self> + ?Sized>(&self, handle: &H, value: <H as Write<Self>>::Value)
    where
        Self: Sized,
    {
        handle.write(self, value)
    }

    /// Create a source cell (`KeepLatest`).
    fn cell<T: PartialEq + 'static>(&self, value: T) -> Source<T>;

    /// Create a source cell — the v2 spelling of [`cell`](Self::cell).
    fn source<T: PartialEq + 'static>(&self, value: T) -> Source<T>;

    /// Create a guarded computed cell.
    fn computed<T, F>(&self, compute: F) -> Computed<T>
    where
        T: PartialEq + 'static,
        F: Fn(&Compute) -> T + 'static;

    /// Create a guarded computed cell with an explicit change predicate.
    fn computed_ripple_when<T, F, C>(&self, compute: F, changed: C) -> Computed<T>
    where
        T: 'static,
        F: Fn(&Compute) -> T + 'static,
        C: Fn(&T, &T) -> bool + 'static;

    /// Create a pass-through (unguarded) derived cell.
    fn slot<T, F>(&self, compute: F) -> Computed<T>
    where
        T: 'static,
        F: Fn(&Compute) -> T + 'static;

    /// Register an effect.
    fn effect<F, R>(&self, run: F) -> Effect
    where
        F: Fn(&Compute) -> R + 'static,
        R: EffectCallbackResult + 'static;

    /// Run `f` inside a batch (coalesced invalidation). The callback receives
    /// the owning [`Context`] — a batch is a write-coalescing scope.
    fn batch<F, R>(&self, f: F) -> R
    where
        F: FnOnce(&Context) -> R;

    /// Tear down a node by id (detach both edge directions, recycle the id).
    fn dispose_node(&self, id: SlotId);
}

impl ComputeOps for Context {
    fn read_value<T: Clone + 'static>(&self, id: SlotId) -> T {
        Context::read_value::<T>(self, id)
    }

    fn get_rc<T: 'static>(&self, handle: &Computed<T>) -> Rc<T> {
        Context::get_rc::<T>(self, handle)
    }

    fn cell<T: PartialEq + 'static>(&self, value: T) -> Source<T> {
        Context::cell::<T>(self, value)
    }

    fn source<T: PartialEq + 'static>(&self, value: T) -> Source<T> {
        Context::source::<T>(self, value)
    }

    fn computed<T, F>(&self, compute: F) -> Computed<T>
    where
        T: PartialEq + 'static,
        F: Fn(&Compute) -> T + 'static,
    {
        Context::computed(self, compute)
    }

    fn computed_ripple_when<T, F, C>(&self, compute: F, changed: C) -> Computed<T>
    where
        T: 'static,
        F: Fn(&Compute) -> T + 'static,
        C: Fn(&T, &T) -> bool + 'static,
    {
        Context::computed_ripple_when(self, compute, changed)
    }

    fn slot<T, F>(&self, compute: F) -> Computed<T>
    where
        T: 'static,
        F: Fn(&Compute) -> T + 'static,
    {
        Context::slot(self, compute)
    }

    fn effect<F, R>(&self, run: F) -> Effect
    where
        F: Fn(&Compute) -> R + 'static,
        R: EffectCallbackResult + 'static,
    {
        Context::effect(self, run)
    }

    fn batch<F, R>(&self, f: F) -> R
    where
        F: FnOnce(&Context) -> R,
    {
        Context::batch(self, f)
    }

    fn dispose_node(&self, id: SlotId) {
        Context::dispose_node(self, id)
    }
}

impl ComputeOps for Compute<'_> {
    fn read_value<T: Clone + 'static>(&self, id: SlotId) -> T {
        self.ctx.register_dependency(id, self.slot_id);
        // Untracked base: the edge is already threaded via `slot_id` above, so
        // the delegate must not register again (would double-register).
        self.ctx.read_value_untracked::<T>(id)
    }

    fn get_rc<T: 'static>(&self, handle: &Computed<T>) -> Rc<T> {
        self.ctx.register_dependency(handle.id, self.slot_id);
        self.ctx.get_rc_untracked(handle)
    }

    fn cell<T: PartialEq + 'static>(&self, value: T) -> Source<T> {
        self.ctx.cell(value)
    }

    fn source<T: PartialEq + 'static>(&self, value: T) -> Source<T> {
        self.ctx.source(value)
    }

    fn computed<T, F>(&self, compute: F) -> Computed<T>
    where
        T: PartialEq + 'static,
        F: Fn(&Compute) -> T + 'static,
    {
        self.ctx.computed(compute)
    }

    fn computed_ripple_when<T, F, C>(&self, compute: F, changed: C) -> Computed<T>
    where
        T: 'static,
        F: Fn(&Compute) -> T + 'static,
        C: Fn(&T, &T) -> bool + 'static,
    {
        self.ctx.computed_ripple_when(compute, changed)
    }

    fn slot<T, F>(&self, compute: F) -> Computed<T>
    where
        T: 'static,
        F: Fn(&Compute) -> T + 'static,
    {
        self.ctx.slot(compute)
    }

    fn effect<F, R>(&self, run: F) -> Effect
    where
        F: Fn(&Compute) -> R + 'static,
        R: EffectCallbackResult + 'static,
    {
        self.ctx.effect(run)
    }

    fn batch<F, R>(&self, f: F) -> R
    where
        F: FnOnce(&Context) -> R,
    {
        self.ctx.batch(f)
    }

    fn dispose_node(&self, id: SlotId) {
        self.ctx.dispose_node(id)
    }
}

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

    // -- #lzspecedgeindex --------------------------------------------------
    //
    // The index side table is only correct while its invariant holds: an owner
    // has an entry exactly while its edge list is longer than the threshold.
    // Both fast paths assert on that, so a violation panics rather than
    // silently reading a stale position.

    #[test]
    fn disposal_invalidates_surviving_readers() {
        // #lzspecedgeindex: detaching the edge is not enough — a reader that
        // still names a disposed node must recompute (and error) rather than
        // serve the value it cached before the disposal, forever.
        let ctx = Context::new();
        let src = ctx.cell(4i64);
        let derived = ctx.computed(move |c| c.get(&src));
        let reader = ctx.computed(move |c| c.get(&derived) + 1);
        assert_eq!(ctx.get(&reader), 5);

        ctx.dispose_slot(&derived);
        let after = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| ctx.get(&reader)));
        assert!(
            after.is_err(),
            "reader must not serve its pre-disposal cache"
        );

        // ... and a later publish on the surviving source must not revive it.
        ctx.set(&src, 99);
        let after_publish =
            std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| ctx.get(&reader)));
        assert!(after_publish.is_err());
    }

    #[test]
    fn degree_accessors_report_live_edge_set_sizes() {
        let ctx = Context::new();
        let topic = ctx.cell(1usize);
        let a = ctx.computed(move |c| c.get(&topic) + 1);
        let b = ctx.computed(move |c| c.get(&a) + 1);
        // Lazy: no edge is registered until the slot is pulled.
        assert_eq!(ctx.dependent_count(&topic), 0);
        assert_eq!(ctx.dependency_count(&b), 0);

        assert_eq!(ctx.get(&b), 3);
        assert_eq!(ctx.dependent_count(&topic), 1);
        assert_eq!(ctx.dependent_count(&a), 1);
        assert_eq!(ctx.dependency_count(&a), 1);
        assert_eq!(ctx.dependency_count(&b), 1);
        // Cells are pure sources; effects are pure sinks.
        assert_eq!(ctx.dependency_count(&topic), 0);
        assert_eq!(ctx.dependent_count(&b), 0);

        // Disposal detaches both directions, and a disposed node reports zero.
        ctx.dispose_slot(&b);
        assert_eq!(ctx.dependent_count(&a), 0);
        assert_eq!(ctx.dependency_count(&b), 0);
    }

    #[test]
    fn degree_accessors_cover_effects() {
        let ctx = Context::new();
        let topic = ctx.cell(1usize);
        let watch = ctx.effect(move |c| {
            let _ = c.get(&topic);
        });
        assert_eq!(ctx.dependency_count(&watch), 1);
        assert_eq!(ctx.dependent_count(&topic), 1);
        assert_eq!(ctx.dependent_count(&watch), 0);

        ctx.dispose_effect(&watch);
        assert_eq!(ctx.dependency_count(&watch), 0);
        assert_eq!(ctx.dependent_count(&topic), 0);
    }

    #[test]
    fn teardown_scope_disposes_its_nodes_on_drop() {
        let ctx = Context::new();
        let topic = ctx.cell(1usize);
        let probe;
        {
            let conn = ctx.scope();
            let a = conn.computed(move |c| c.get(&topic) + 1);
            let _b = conn.computed(move |c| c.get(&a) * 10);
            assert_eq!(conn.len(), 2);
            probe = a;
            assert_eq!(ctx.get(&probe), 2);
            match Context::get_node(&ctx.inner.borrow().nodes, topic.id) {
                Some(Node::Source(c)) => assert!(c.dependents.contains(&probe.id)),
                _ => panic!("expected cell"),
            }
        }
        let inner = ctx.inner.borrow();
        match Context::get_node(&inner.nodes, topic.id) {
            Some(Node::Source(c)) => assert!(
                c.dependents.is_empty(),
                "the scope's edges must be gone with it"
            ),
            _ => panic!("expected cell"),
        }
        assert!(inner.free_ids.contains(&probe.id.0), "ids recycled");
    }

    #[test]
    fn teardown_scope_keeps_handles_copy() {
        // The point of the design: a source is captured by two closures with no
        // clone, because handles are still Copy ids.
        let ctx = Context::new();
        let topic = ctx.cell(5usize);
        let conn = ctx.scope();
        let a = conn.computed(move |c| c.get(&topic) + 1);
        let b = conn.computed(move |c| c.get(&topic) + 2);
        assert_eq!(ctx.get(&a) + ctx.get(&b), 13);
    }

    #[test]
    fn teardown_scope_reads_across_scopes() {
        // Grouping bounds teardown, not visibility.
        let ctx = Context::new();
        let topic = ctx.cell(2usize);
        let outer = ctx.computed(move |c| c.get(&topic) * 3);
        let conn = ctx.scope();
        let inner = conn.computed(move |c| c.get(&outer) + 1);
        assert_eq!(ctx.get(&inner), 7);
    }

    #[test]
    fn teardown_scope_owns_cells_and_effects_too() {
        let ctx = Context::new();
        let (cell_id, effect_id);
        {
            let conn = ctx.scope();
            let cell = conn.source(1usize);
            cell_id = cell.id;
            let effect = conn.effect(move |c| {
                let _ = c.get(&cell);
            });
            effect_id = effect.id;
            assert_eq!(conn.len(), 2);
        }
        let inner = ctx.inner.borrow();
        assert!(inner.free_ids.contains(&cell_id.0), "cell recycled");
        assert!(inner.free_ids.contains(&effect_id.0), "effect recycled");
    }

    #[test]
    fn teardown_scope_disarm_keeps_the_nodes_alive() {
        let ctx = Context::new();
        let topic = ctx.cell(1usize);
        let escaped = {
            let conn = ctx.scope();
            let slot = conn.computed(move |c| c.get(&topic) * 10);
            conn.disarm();
            slot
        };
        assert_eq!(ctx.get(&escaped), 10);
        ctx.set(&topic, 4);
        assert_eq!(ctx.get(&escaped), 40);
    }

    #[test]
    fn teardown_scope_keeps_churn_flat() {
        // The churn case, written as scopes: each subscriber is its own scope.
        let ctx = Context::new();
        let topic = ctx.cell(0usize);
        for cycle in 0..500usize {
            let conn = ctx.scope();
            let slot = conn.computed(move |c| c.get(&topic) + cycle);
            assert_eq!(ctx.get(&slot), cycle);
        }
        match Context::get_node(&ctx.inner.borrow().nodes, topic.id) {
            Some(Node::Source(c)) => assert!(
                c.dependents.is_empty(),
                "every scope must have torn itself down"
            ),
            _ => panic!("expected cell"),
        }
    }

    #[test]
    fn disposal_tolerates_a_capture_whose_drop_reenters() {
        // A node owns its compute closure and everything that closure captured.
        // If the node is dropped while the RefCell borrow is held, any capture
        // whose Drop touches the context panics with "already borrowed" — which
        // a self-disposing handle type does by construction.
        struct ReentersOnDrop {
            ctx: std::rc::Weak<Context>,
            victim: SlotId,
        }
        impl Drop for ReentersOnDrop {
            fn drop(&mut self) {
                if let Some(ctx) = self.ctx.upgrade() {
                    // any context call that borrows would do
                    ctx.dispose_slot(&Computed::<u64> {
                        id: self.victim,
                        _marker: std::marker::PhantomData,
                    });
                }
            }
        }

        let ctx = std::rc::Rc::new(Context::new());
        let base = ctx.cell(1u64);
        let victim = ctx.computed(move |c| c.get(&base) + 1);
        assert_eq!(ctx.get(&victim), 2);

        let reentrant = ReentersOnDrop {
            ctx: std::rc::Rc::downgrade(&ctx),
            victim: victim.id,
        };
        let holder = ctx.computed(move |c| {
            let _ = &reentrant;
            c.get(&base) + 100
        });
        assert_eq!(ctx.get(&holder), 101);

        // Disposing `holder` drops its closure, dropping `reentrant`, whose Drop
        // disposes `victim`. Must not panic.
        ctx.dispose_slot(&holder);
        assert!(
            ctx.inner.borrow().free_ids.contains(&victim.id.0),
            "the re-entrant disposal must have completed"
        );
    }

    #[test]
    fn dispose_slot_detaches_both_edge_directions_and_recycles_the_id() {
        let ctx = Context::new();
        let src = ctx.cell(1usize);
        let mid = ctx.computed(move |ctx| ctx.get(&src) + 1);
        let sink = ctx.computed(move |ctx| ctx.get(&mid) * 10);
        assert_eq!(ctx.get(&sink), 20);

        let recycled = mid.id;
        ctx.dispose_slot(&mid);

        {
            let inner = ctx.inner.borrow();
            // upstream: the cell no longer lists the disposed slot
            match Context::get_node(&inner.nodes, src.id) {
                Some(Node::Source(c)) => assert!(
                    !c.dependents.contains(&recycled),
                    "dependency must drop the disposed dependent"
                ),
                _ => panic!("expected cell"),
            }
            // downstream: the sink no longer lists the disposed slot
            match Context::get_node(&inner.nodes, sink.id) {
                Some(Node::Computed(s)) => assert!(
                    !s.dependencies.contains(&recycled),
                    "dependent must drop the disposed dependency"
                ),
                _ => panic!("expected slot"),
            }
            assert!(inner.free_ids.contains(&recycled.0), "id must be recycled");
        }

        // the recycled id is handed to the next node and behaves normally
        let reused = ctx.computed(move |ctx| ctx.get(&src) + 100);
        assert_eq!(reused.id, recycled);
        assert_eq!(ctx.get(&reused), 101);
        ctx.set(&src, 5);
        assert_eq!(ctx.get(&reused), 105);
    }

    #[test]
    fn dispose_cell_detaches_dependents_and_recycles_the_id() {
        let ctx = Context::new();
        let src = ctx.cell(2usize);
        let derived = ctx.computed(move |ctx| ctx.get(&src) * 3);
        assert_eq!(ctx.get(&derived), 6);

        let recycled = src.id;
        ctx.dispose_cell(&src);

        let inner = ctx.inner.borrow();
        match Context::get_node(&inner.nodes, derived.id) {
            Some(Node::Computed(s)) => assert!(
                !s.dependencies.contains(&recycled),
                "dependent must drop the disposed cell"
            ),
            _ => panic!("expected slot"),
        }
        assert!(inner.free_ids.contains(&recycled.0), "id must be recycled");
    }

    #[test]
    fn dispose_slot_ignores_a_handle_naming_another_kind() {
        // free_ids is LIFO, so a stale handle can name a live node of a
        // different kind. Disposing through it must be a no-op, not a teardown.
        let ctx = Context::new();
        let cell = ctx.cell(1usize);
        let stale: Computed<usize> = Computed {
            id: cell.id,
            _marker: std::marker::PhantomData,
        };
        ctx.dispose_slot(&stale);
        assert_eq!(ctx.get(&cell), 1, "the cell must survive");
        assert!(
            !ctx.inner.borrow().free_ids.contains(&cell.id.0),
            "a live node's id must not be recycled"
        );
    }

    #[test]
    fn dispose_slot_returns_the_graph_to_its_prior_size() {
        // The churn case: repeatedly subscribe and unsubscribe against one
        // topic. Without disposal the dependent list grows without bound even
        // though the live count is constant.
        let ctx = Context::new();
        let topic = ctx.cell(0usize);
        let live_width = 8usize;
        let mut live_a: Vec<_> = (0..live_width)
            .map(|i| {
                let slot = ctx.computed(move |ctx| ctx.get(&topic) + i);
                ctx.get(&slot);
                slot
            })
            .collect();

        for cycle in 0..500usize {
            let victim = live_a.swap_remove(cycle % live_a.len());
            ctx.dispose_slot(&victim);
            let slot = ctx.computed(move |ctx| ctx.get(&topic) + cycle);
            ctx.get(&slot);
            live_a.push(slot);
        }

        let inner = ctx.inner.borrow();
        match Context::get_node(&inner.nodes, topic.id) {
            Some(Node::Source(c)) => assert_eq!(
                c.dependents.len(),
                live_width,
                "dependent list must track live subscribers, not total ever created"
            ),
            _ => panic!("expected cell"),
        }
    }

    #[test]
    fn edge_index_promotes_past_the_threshold_and_still_dedups() {
        let ctx = Context::new();
        let src = ctx.cell(0usize);
        // comfortably past EDGE_INDEX_THRESHOLD so the index is built
        let width = EDGE_INDEX_THRESHOLD * 4;
        let dep_a: Vec<_> = (0..width)
            .map(|i| ctx.computed(move |ctx| ctx.get(&src) + i))
            .collect();
        for slot in &dep_a {
            ctx.get(slot);
        }

        {
            let inner = ctx.inner.borrow();
            let index = inner
                .dependents_index
                .get(&src.id)
                .expect("wide cell must be indexed");
            assert_eq!(index.len(), width, "one index entry per dependent");
        }

        // re-reading must not duplicate edges
        for slot in &dep_a {
            ctx.get(slot);
        }
        {
            let inner = ctx.inner.borrow();
            match Context::get_node(&inner.nodes, src.id) {
                Some(Node::Source(c)) => {
                    assert_eq!(c.dependents.len(), width, "no duplicate edges")
                }
                _ => panic!("expected cell"),
            }
        }

        ctx.set(&src, 7);
        for (i, slot) in dep_a.iter().enumerate() {
            assert_eq!(ctx.get(slot), 7 + i, "every dependent recomputed");
        }
    }

    #[test]
    fn edge_index_demotes_when_the_list_shrinks_below_the_threshold() {
        let ctx = Context::new();
        let src = ctx.cell(0usize);
        let toggle = ctx.cell(true);
        // each slot reads src only while `toggle` is set, so flipping it drops
        // every dependent edge and must demote the index
        let width = EDGE_INDEX_THRESHOLD * 2;
        let dep_a: Vec<_> = (0..width)
            .map(|i| {
                ctx.computed(move |ctx| {
                    if ctx.get(&toggle) {
                        ctx.get(&src) + i
                    } else {
                        i
                    }
                })
            })
            .collect();
        for slot in &dep_a {
            ctx.get(slot);
        }
        assert!(
            ctx.inner.borrow().dependents_index.contains_key(&src.id),
            "wide list must be indexed"
        );

        ctx.set(&toggle, false);
        for slot in &dep_a {
            ctx.get(slot);
        }
        assert!(
            !ctx.inner.borrow().dependents_index.contains_key(&src.id),
            "index entry must be dropped once the list is short again"
        );

        // and the graph still behaves
        ctx.set(&src, 99);
        for (i, slot) in dep_a.iter().enumerate() {
            assert_eq!(ctx.get(slot), i, "src is no longer a dependency");
        }
    }

    #[test]
    fn edge_index_does_not_survive_id_recycling() {
        // free_ids is LIFO, so a disposed effect's id is handed straight to the
        // next node. A leftover index entry would alias it.
        let ctx = Context::new();
        let src = ctx.cell(0usize);
        let width = EDGE_INDEX_THRESHOLD * 2;
        let cell_a: Vec<_> = (0..width).map(|i| ctx.cell(i)).collect();
        let effect = ctx.effect(move |ctx| {
            for cell in &cell_a {
                let _ = ctx.get(cell);
            }
        });
        assert!(
            ctx.inner
                .borrow()
                .dependencies_index
                .contains_key(&effect.id),
            "wide effect must be indexed"
        );

        let recycled = effect.id;
        ctx.dispose_effect(&effect);
        assert!(
            !ctx.inner
                .borrow()
                .dependencies_index
                .contains_key(&recycled),
            "disposal must drop the index entry before the id is recycled"
        );

        let reused = ctx.computed(move |ctx| ctx.get(&src) + 1);
        assert_eq!(reused.id, recycled, "id was recycled as expected");
        assert_eq!(ctx.get(&reused), 1, "recycled node computes normally");
        ctx.set(&src, 41);
        assert_eq!(ctx.get(&reused), 42, "recycled node propagates normally");
    }

    #[test]
    fn context_nodes_are_vec_indexed_by_sequential_slot_ids_and_reuse_effect_ids() {
        let ctx = Context::new();
        let cell = ctx.cell(1i32);
        let slot = ctx.slot(|_| 2i32);
        let effect = ctx.effect(move |ctx| {
            let _ = ctx.get(&slot);
        });

        assert_eq!(cell.id, SlotId(0));
        assert_eq!(slot.id, SlotId(1));
        assert_eq!(effect.id, SlotId(2));
        {
            let inner = ctx.inner.borrow();
            assert_eq!(inner.nodes.len(), 3);
            assert_eq!(inner.next_id, 3);
            assert!(inner.free_ids.is_empty());
            assert!(matches!(inner.nodes[0].as_ref(), Some(Node::Source(_))));
            assert!(matches!(inner.nodes[1].as_ref(), Some(Node::Computed(_))));
            assert!(matches!(inner.nodes[2].as_ref(), Some(Node::Effect(_))));
        }

        effect.dispose(&ctx);
        {
            let inner = ctx.inner.borrow();
            assert_eq!(inner.nodes.len(), 3);
            assert_eq!(inner.next_id, 3);
            assert_eq!(inner.free_ids.as_slice(), &[2]);
            assert!(inner.nodes[2].is_none());
        }

        let reused = ctx.computed(|_| 3i32);
        assert_eq!(reused.id, SlotId(2));
        {
            let inner = ctx.inner.borrow();
            assert_eq!(inner.nodes.len(), 3);
            assert_eq!(inner.next_id, 3);
            assert!(inner.free_ids.is_empty());
            assert!(matches!(inner.nodes[2].as_ref(), Some(Node::Computed(_))));
        }
    }

    /// #lzspecedgeindex: `remove_pending_effect` skips the O(queue) scan when
    /// the scheduled flag is clear, which is sound only because a set flag
    /// implies queue membership. Pin that invariant — if `schedule_effect` ever
    /// sets the flag without pushing, the skip would silently leak a queued id.
    #[test]
    fn a_set_scheduled_flag_implies_queue_membership() {
        let ctx = Context::new();
        let cell = ctx.cell(0i32);
        let effect_a: Vec<_> = (0..8)
            .map(|_| {
                ctx.effect(move |ctx| {
                    let _ = ctx.get(&cell);
                })
            })
            .collect();

        // Re-scheduling an already-scheduled effect must not double-push.
        for effect in &effect_a {
            ctx.schedule_effect(effect.id, true);
            ctx.schedule_effect(effect.id, true);
        }

        let inner = ctx.inner.borrow();
        for effect in &effect_a {
            let idx = Context::node_index(effect.id).unwrap();
            if inner.scheduled_effects[idx] {
                assert_eq!(
                    inner
                        .pending_effects
                        .iter()
                        .filter(|queued| **queued == effect.id)
                        .count(),
                    1,
                    "a set scheduled flag must mean exactly one queue entry"
                );
            }
        }
    }

    /// #lzspecedgeindex: the flush-path scan removal must not drop a run. Every
    /// effect on a wide topic still fires exactly once per publish.
    #[test]
    fn wide_fan_out_effects_each_run_once_per_publish() {
        use std::cell::RefCell;
        use std::rc::Rc;

        const WIDTH: usize = 64;
        let ctx = Context::new();
        let cell = ctx.cell(0i32);
        let run_count = Rc::new(RefCell::new(vec![0usize; WIDTH]));

        for i in 0..WIDTH {
            let run_count = Rc::clone(&run_count);
            ctx.effect(move |ctx| {
                let _ = ctx.get(&cell);
                run_count.borrow_mut()[i] += 1;
            });
        }
        // Creation runs each effect once.
        assert_eq!(*run_count.borrow(), vec![1usize; WIDTH]);

        ctx.set(&cell, 1);
        assert_eq!(*run_count.borrow(), vec![2usize; WIDTH]);

        ctx.set(&cell, 2);
        assert_eq!(*run_count.borrow(), vec![3usize; WIDTH]);
    }

    #[test]
    fn dispose_effect_deschedules_without_scanning_the_queue() {
        // #lzspecedgeindex: dispose no longer drains the queue eagerly — that
        // scan was O(queue) per disposal, and a mass teardown during a flush
        // made it O(W^2). It clears the scheduled flag instead and leaves a
        // tombstone, which `pop_scheduled_effect` discards. What must hold is
        // the *flag*, not the queue entry.
        let ctx = Context::new();
        let cell = ctx.cell(0i32);
        let effect = ctx.effect(move |ctx| {
            let _ = ctx.get(&cell);
        });

        // Simulate the effect being scheduled (pending) but not yet flushed,
        // which happens when another effect's body invalidates it mid-flush.
        ctx.schedule_effect(effect.id, true);
        {
            let inner = ctx.inner.borrow();
            assert!(inner.pending_effects.contains(&effect.id));
        }
        assert!(ctx.is_effect_scheduled(effect.id));

        effect.dispose(&ctx);

        let inner = ctx.inner.borrow();
        assert!(
            !ctx.is_effect_scheduled(effect.id),
            "dispose must clear the scheduled flag"
        );
        assert_eq!(inner.free_ids.as_slice(), &[effect.id.0]);
    }

    /// #lzspecedgeindex: the hazard the old eager queue scan existed to
    /// prevent. Disposal leaves a tombstone and frees the id; `free_ids` is
    /// LIFO, so the very next node allocated reuses it and now aliases that
    /// tombstone. The recycled node must still run exactly once — never twice
    /// (tombstone plus its own entry) and never zero times.
    #[test]
    fn a_tombstone_does_not_disturb_a_recycled_id() {
        use std::cell::RefCell;
        use std::rc::Rc;

        let ctx = Context::new();
        let cell = ctx.cell(0i32);

        let doomed = ctx.effect(move |ctx| {
            let _ = ctx.get(&cell);
        });
        // Queue it, then dispose it: entry stays behind, id goes on free_ids.
        ctx.schedule_effect(doomed.id, true);
        doomed.dispose(&ctx);

        // The next effect recycles that exact id, aliasing the tombstone.
        let run_count = Rc::new(RefCell::new(0usize));
        let recycled = {
            let run_count = Rc::clone(&run_count);
            ctx.effect(move |ctx| {
                let _ = ctx.get(&cell);
                *run_count.borrow_mut() += 1;
            })
        };
        assert_eq!(recycled.id, doomed.id, "id must actually be recycled");
        assert_eq!(*run_count.borrow(), 1, "creation runs it once");

        // A publish must run it exactly once despite the aliased tombstone.
        ctx.set(&cell, 1);
        assert_eq!(
            *run_count.borrow(),
            2,
            "recycled effect must run exactly once per publish"
        );

        // And the tombstone must not resurrect it after a real disposal.
        recycled.dispose(&ctx);
        ctx.set(&cell, 2);
        assert_eq!(
            *run_count.borrow(),
            2,
            "a disposed effect must never run again"
        );
    }

    // -- Cycle detection (#lzcycledetect) ----------------------------------

    fn cycle_panic_message(result: Box<dyn std::any::Any + Send>) -> String {
        result
            .downcast_ref::<String>()
            .cloned()
            .or_else(|| result.downcast_ref::<&str>().map(|s| s.to_string()))
            .unwrap_or_default()
    }

    #[test]
    fn two_slot_dependency_cycle_panics_instead_of_overflowing() {
        let ctx = Context::new();
        // `a` reads `b` (wired after both exist); `b` reads `a`.
        let link: std::rc::Rc<std::cell::RefCell<Option<Computed<i32>>>> =
            std::rc::Rc::new(std::cell::RefCell::new(None));
        let link_a = std::rc::Rc::clone(&link);
        let a = ctx.computed(move |ctx| match *link_a.borrow() {
            Some(b) => ctx.get(&b) + 1,
            None => 0,
        });
        let b = ctx.computed(move |ctx| ctx.get(&a) + 1);
        *link.borrow_mut() = Some(b);

        let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| ctx.get(&a)));
        assert!(
            result.is_err(),
            "circular dependency must panic, not diverge"
        );
        let msg = cycle_panic_message(result.unwrap_err());
        assert!(
            msg.contains("circular dependency"),
            "unexpected panic message: {msg}"
        );

        // The in-progress flags must be cleared by the RAII guards so the
        // context is not wedged after the panic is caught.
        {
            let inner = ctx.inner.borrow();
            for node in inner.nodes.iter().flatten() {
                if let Node::Computed(slot) = node {
                    assert!(!slot.in_progress, "in_progress must be cleared on unwind");
                }
            }
        }
        let fresh = ctx.computed(|_| 42i32);
        assert_eq!(ctx.get(&fresh), 42, "context must still work after a cycle");
    }

    #[test]
    fn self_referential_slot_panics() {
        let ctx = Context::new();
        let link: std::rc::Rc<std::cell::RefCell<Option<Computed<i32>>>> =
            std::rc::Rc::new(std::cell::RefCell::new(None));
        let link_self = std::rc::Rc::clone(&link);
        let s = ctx.computed(move |ctx| match *link_self.borrow() {
            Some(me) => ctx.get(&me) + 1,
            None => 0,
        });
        *link.borrow_mut() = Some(s);

        let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| ctx.get(&s)));
        assert!(result.is_err(), "self-referential slot must panic");
        let msg = cycle_panic_message(result.unwrap_err());
        assert!(msg.contains("circular dependency"), "got: {msg}");
    }

    #[test]
    fn diamond_dependencies_do_not_false_positive() {
        // A diamond (`d` <- b,c <- a) reads `a` twice but is acyclic; the
        // cycle guard must not misfire on shared, non-nested dependencies.
        let ctx = Context::new();
        let a = ctx.cell(1i32);
        let b = ctx.computed(move |ctx| ctx.get(&a) + 1);
        let c = ctx.computed(move |ctx| ctx.get(&a) + 2);
        let d = ctx.computed(move |ctx| ctx.get(&b) + ctx.get(&c));
        assert_eq!(ctx.get(&d), (1 + 1) + (1 + 2));
        a.set(&ctx, 10);
        assert_eq!(ctx.get(&d), (10 + 1) + (10 + 2));
    }

    #[test]
    fn batch_dedup_invalidates_shared_dependent_once(/* #lzbatchalloc */) {
        // Multiple writes to the same cell inside one batch, plus writes to
        // several cells sharing one dependent, must coalesce: the shared
        // dependent is invalidated exactly once on flush. Guards the sort+dedup
        // replacement for the former HashSet-backed batch sets.
        let ctx = Context::new();
        let a = ctx.cell(0i32);
        let b = ctx.cell(0i32);
        let total = ctx.computed(move |ctx| ctx.get(&a) + ctx.get(&b));
        assert_eq!(ctx.get(&total), 0);

        ctx.batch(|ctx| {
            a.set(ctx, 1);
            a.set(ctx, 2); // duplicate cell `a` in the batch set
            a.set(ctx, 3); // triplicate
            b.set(ctx, 10);
        });
        // After the batch flush, the dependent sees the latest coalesced values.
        assert_eq!(ctx.get(&total), 13);
    }
}

#[cfg(audit_probe)]
pub mod audit_probe {
    use std::cell::Cell as StdCell;

    thread_local! {
        static MAX_LEN: StdCell<usize> = const { StdCell::new(0) };
        static TOTAL_LEN: StdCell<u64> = const { StdCell::new(0) };
        static CALLS: StdCell<u64> = const { StdCell::new(0) };
    }

    pub(crate) fn record_dispose_queue_len(len: usize) {
        MAX_LEN.with(|m| m.set(m.get().max(len)));
        TOTAL_LEN.with(|t| t.set(t.get() + len as u64));
        CALLS.with(|c| c.set(c.get() + 1));
    }

    /// Static layout of every type that participates in one reactive node
    /// (#lzspecedgeindex, step 1 of the per-subscriber memory audit).
    ///
    /// Returns `(name, size_of, align_of)`. These are the load-immune half of
    /// the per-node memory question: what the arena costs before a single byte
    /// is allocated. `examples/node_memory_audit.rs` pairs them with counted
    /// allocations so the two halves can be summed against a measured total.
    pub fn layout_rows() -> Vec<(&'static str, usize, usize)> {
        use core::mem::{align_of, size_of};
        macro_rules! row {
            ($t:ty) => {
                (stringify!($t), size_of::<$t>(), align_of::<$t>())
            };
        }
        vec![
            row!(super::SlotId),
            row!(super::TypeTag),
            row!(super::InlineBuf),
            row!(super::AnyValue),
            row!(super::EdgeVec),
            row!(std::rc::Rc<super::ComputeFn>),
            row!(Option<Box<super::EqualsFn>>),
            row!(std::rc::Rc<super::EffectFn>),
            row!(Option<Box<dyn FnOnce()>>),
            row!(super::ComputedNode),
            row!(super::SourceNode),
            row!(super::EffectNode),
            row!(super::Node),
            row!(Option<super::Node>),
        ]
    }

    /// Byte offset of each `ComputedNode` field, so the padding and the
    /// largest-variant tax in `Node` can be attributed rather than guessed.
    pub fn slot_node_field_offsets() -> Vec<(&'static str, usize)> {
        use core::mem::offset_of;
        vec![
            ("value", offset_of!(super::ComputedNode, value)),
            ("type_id", offset_of!(super::ComputedNode, type_id)),
            ("compute", offset_of!(super::ComputedNode, compute)),
            ("equals", offset_of!(super::ComputedNode, equals)),
            (
                "dependencies",
                offset_of!(super::ComputedNode, dependencies),
            ),
            ("dependents", offset_of!(super::ComputedNode, dependents)),
            ("dirty", offset_of!(super::ComputedNode, dirty)),
            (
                "force_recompute",
                offset_of!(super::ComputedNode, force_recompute),
            ),
            ("in_progress", offset_of!(super::ComputedNode, in_progress)),
            ("verified_at", offset_of!(super::ComputedNode, verified_at)),
        ]
    }

    /// (max queue len seen, mean queue len, dispose_effect calls)
    pub fn take() -> (usize, f64, u64) {
        let max = MAX_LEN.with(|m| m.replace(0));
        let total = TOTAL_LEN.with(|t| t.replace(0));
        let calls = CALLS.with(|c| c.replace(0));
        let mean = if calls == 0 {
            0.0
        } else {
            total as f64 / calls as f64
        };
        (max, mean, calls)
    }
}