graphrefly_core/

node.rs

//! The dispatcher — node registration, subscription, wave engine.
//!
//! Mirrors `~/src/graphrefly-ts/src/__experiments__/handle-core/core.ts`
//! (the Phase 13.6 brainstorm prototype, ~370 lines, 22 invariant tests).
//!
//! # Scope (M1 dispatcher + Slice A+B parity, closed 2026-05-05)
//!
//! - State + derived + dynamic node registration.
//! - Subscribe / unsubscribe with push-on-subscribe (R1.2.3).
//! - RAII [`Subscription`] with Drop-based deregister (§10.12).
//! - DIRTY → DATA / RESOLVED ordering (R1.3.1.b two-phase push).
//! - Equals-substitution (R1.3.2): identity is zero-FFI; custom crosses boundary.
//! - First-run gate (R2.5.3) — fn does not fire until every dep has a handle.
//! - Diamond resolution — one fn fire per wave even with shared upstream.
//! - `set_deps()` atomic dep mutation with cycle detection + Phase 13.8 Q1
//!   terminal-rejection policy (R3.3.1).
//! - PAUSE / RESUME with lockId set + replay buffer (R1.2.6, R2.6, §10.2).
//! - INVALIDATE broadcast + cascade with R1.4 idempotency.
//! - COMPLETE / ERROR cascade + Lock 2.B auto-cascade gating
//!   (ERROR dominates COMPLETE; first error wins).
//! - TEARDOWN auto-precedes COMPLETE (R2.6.4 / Lock 6.F) +
//!   `has_received_teardown` idempotency.
//! - Meta TEARDOWN ordering (R1.3.9.d) — companions tear down before parent.
//! - Resubscribable terminal lifecycle (R2.2.7, R2.5.3) — late subscribe to a
//!   resubscribable terminal node resets lifecycle, except after TEARDOWN
//!   (per F3 audit guard: TEARDOWN is permanent).
//!
//! # Module split (Slice C-1, 2026-05-05)
//!
//! Wave-engine internals (drain loop, fire selection, emission commit, sink
//! dispatch) live in [`crate::batch`]. The split is purely organizational —
//! the methods are still on `Core`. See `batch.rs` for the wave-engine
//! entry points (`run_wave`, `drain_and_flush`, `commit_emission`,
//! `queue_notify`, `deliver_data_to_consumer`).
//!
//! # Out of scope (later slices / milestones)
//!
//! - Deactivation cleanup (RAM nodes clear cache when sink count → 0) — M2.
//!
//! See [`migration-status.md`](../../../docs/migration-status.md) for the
//! milestone tracker and [`porting-deferred.md`](../../../docs/porting-deferred.md)
//! for surfaced concerns deferred to evidence-driven slices.
//!
//! # Re-entrance discipline (Slice A close, M1: fully lock-released)
//!
//! - **Wave-end sink fires** drop the state lock first. A subscriber's sink
//!   that calls back into `Core::emit` / `pause` / `resume` / `invalidate` /
//!   `complete` / `error` / `teardown` re-acquires the lock cleanly and runs
//!   a nested wave (`s.in_tick` is cleared before the deferred-fire phase).
//! - **`BindingBoundary::invoke_fn`** fires lock-released. The wave engine
//!   acquires + drops the state lock per fn-fire iteration around the
//!   `invoke_fn` callback. User fns may re-enter `Core::emit` / `pause` /
//!   etc. and run a nested wave.
//! - **`BindingBoundary::custom_equals`** fires lock-released.
//!   `commit_emission` brackets the equals check around a lock release;
//!   custom equals oracles may re-enter Core safely.
//! - **Subscribe-time handshake** also fires lock-released. [`Core::subscribe`]
//!   acquires the [`Core::wave_owner`] re-entrant mutex first (cross-thread
//!   serialization), installs the sink under the state lock, drops the state
//!   lock, then fires the per-tier handshake (`[Start]` / `[Data(cache)]?` /
//!   `[Complete]?` / `[Error(h)]?` / `[Teardown]?` per R1.3.5.a) lock-released.
//!   A handshake-time sink callback may re-enter Core (`emit` / `complete` /
//!   `error` / `subscribe`); same-thread re-entry passes through `wave_owner`
//!   transparently. Cross-thread emits block on `wave_owner` until the
//!   subscribe path drops it, preserving R1.3.5.a happens-after ordering.

use std::collections::VecDeque;
use std::panic::{catch_unwind, AssertUnwindSafe};
use std::sync::{Arc, Weak};

use ahash::{AHashMap as HashMap, AHashSet as HashSet};
use indexmap::IndexMap;
use parking_lot::{Mutex, MutexGuard, ReentrantMutex};
use smallvec::SmallVec;
use thiserror::Error;

use crate::batch::PendingPerNode;
use crate::boundary::{BindingBoundary, CleanupTrigger};
use crate::clock::monotonic_ns;
use crate::handle::{FnId, HandleId, LockId, NodeId, NO_HANDLE};
use crate::message::Message;

/// Terminal-lifecycle state — once set on a node, the node will not emit
/// further DATA; per-dep slots on consumers also use this to track which
/// upstreams have terminated (R1.3.4 / Lock 2.B).
///
/// `Error` carries a [`HandleId`] resolving to the error value. Refcount is
/// retained when the variant is stored in a node's `terminal` slot or any
/// consumer's `dep_terminals` slot; v1 does not release these (terminal
/// state is one-shot at this layer; release happens on resubscribable
/// terminal-lifecycle reset, a separate slice).
#[derive(Copy, Clone, Debug, PartialEq, Eq)]
pub enum TerminalKind {
    Complete,
    Error(HandleId),
}

/// Node kind discriminant — **derived metadata** computed from
/// [`NodeRecord`]'s field shape (D030 unification, Slice D).
///
/// Core no longer stores `kind` as a field; it's computed on demand from
/// `(deps.is_empty(), fn_id.is_some(), op.is_some(), is_dynamic)`,
/// mirroring TS's data model where `NodeImpl` has no `_kind` field. The
/// shape uniquely identifies the kind:
///
/// | deps      | fn_id | op   | is_dynamic | kind     |
/// |-----------|-------|------|-----------|----------|
/// | empty     | None  | None | -         | State    |
/// | empty     | Some  | None | -         | Producer |
/// | non-empty | Some  | None | false     | Derived  |
/// | non-empty | Some  | None | true      | Dynamic  |
/// | non-empty | None  | Some | -         | Operator |
///
/// Public API ([`Core::kind_of`]) derives this enum on each call. State
/// nodes are ROM (cache survives deactivation); compute nodes
/// (Derived / Dynamic / Operator) and producers are RAM.
#[derive(Copy, Clone, Eq, PartialEq, Debug)]
pub enum NodeKind {
    /// Source node: cache is intrinsic, no fn, no deps. Mutated via [`Core::emit`].
    State,
    /// Producer node: fn fires once on first subscribe. No deps;
    /// emissions arrive via sinks the fn subscribes to (zip / concat /
    /// race / takeUntil pattern). Slice D / D031.
    Producer,
    /// Derived node: fn fires on every dep change; all deps tracked.
    Derived,
    /// Dynamic node: fn declares which dep indices it actually read this run.
    /// Untracked dep updates flow through cache but do NOT re-fire fn.
    Dynamic,
    /// Operator node: built-in dispatch path for transform / combine /
    /// flow / resilience operators. The `OperatorOp` discriminant selects
    /// the per-operator FFI path ([`BindingBoundary::project_each`] etc.);
    /// Core manages per-operator state via the generic `op_scratch` slot
    /// on `NodeRecord` (D026). Per Slice C-1 (D009) / Slice C-3 (D026).
    Operator(OperatorOp),
}

impl NodeKind {
    /// True if this kind opts OUT of Lock 2.B auto-cascade. Operator(Reduce)
    /// and Operator(Last) must intercept upstream COMPLETE so they can emit
    /// their accumulator / buffered value before the cascade terminates them;
    /// instead of cascading, terminate_node queues such children for fn-fire
    /// so `fire_operator` can handle the terminal.
    pub(crate) fn skips_auto_cascade(self) -> bool {
        matches!(
            self,
            NodeKind::Operator(OperatorOp::Reduce { .. } | OperatorOp::Last { .. })
        )
    }
}

/// Built-in operator discriminant. Selects the per-operator dispatch path
/// in `fire_operator` (`crates/graphrefly-core/src/batch.rs`). Each variant
/// carries the binding-side closure ids (and seed handle for stateful
/// folders) needed for the wave-execution path; Core stores no user values
/// itself per the handle-protocol cleaving plane.
#[derive(Copy, Clone, Eq, PartialEq, Debug)]
pub enum OperatorOp {
    /// `map(source, project)` — element-wise transform. Calls
    /// `BindingBoundary::project_each(fn_id, &inputs)` per fire; emits each
    /// returned handle via `commit_emission_verbatim` (R1.3.2.d batch
    /// semantics — no equals substitution between batch entries).
    Map { fn_id: FnId },
    /// `filter(source, predicate)` — silent-drop selection (D012/D018).
    /// Calls `BindingBoundary::predicate_each(fn_id, &inputs)`; emits each
    /// passing input verbatim. If zero pass on a wave that dirtied the
    /// node, queues a single `RESOLVED` to settle (D018).
    Filter { fn_id: FnId },
    /// `scan(source, fold, seed)` — left-fold emitting each new accumulator.
    /// `seed` is captured at registration; `acc` lives in
    /// [`ScanState`](super::op_state::ScanState) inside
    /// [`NodeRecord::op_scratch`] and persists across waves until
    /// resubscribable reset. Calls `BindingBoundary::fold_each(fn_id, acc,
    /// &inputs) -> SmallVec<HandleId>` per fire.
    Scan { fn_id: FnId, seed: HandleId },
    /// `reduce(source, fold, seed)` — left-fold emitting once on upstream
    /// COMPLETE. Accumulates silently while source DATA flows; on
    /// dep[0].terminal == Some(Complete), emits `[Data(acc), Complete]`.
    /// On `Error(h)`, propagates the error verbatim. Opts out of Lock 2.B
    /// auto-cascade (see `NodeKind::skips_auto_cascade`).
    Reduce { fn_id: FnId, seed: HandleId },
    /// `distinctUntilChanged(source, equals)` — suppresses adjacent
    /// duplicates. Calls `BindingBoundary::custom_equals(equals_fn_id,
    /// prev, current)` per input; emits non-equal items verbatim and
    /// updates `prev`. If zero items pass on a wave that dirtied the node,
    /// queues `RESOLVED` (matches Filter discipline).
    DistinctUntilChanged { equals_fn_id: FnId },
    /// `pairwise(source)` — emits `(prev, current)` pairs starting after
    /// the second value. First value swallowed (sets `prev`). Calls
    /// `BindingBoundary::pairwise_pack(fn_id, prev, current)` per pair to
    /// produce the binding-side tuple handle.
    Pairwise { fn_id: FnId },

    // ----- Slice C-2: multi-dep combinators (D020) -----
    /// `combine(...sources)` — N-dep combineLatest. On any dep fire, packs
    /// the latest handle per dep into a single tuple handle via
    /// `BindingBoundary::pack_tuple(pack_fn, &handles)`. First-run gate
    /// (`partial: false` default) holds until all deps deliver real DATA
    /// (R2.5.3). COMPLETE cascades when all deps complete (R1.3.4.b).
    Combine { pack_fn: FnId },

    /// `withLatestFrom(primary, secondary)` — 2-dep, fire-on-primary-only
    /// (D021, Phase 10.5). Packs `[primary, secondary]` via
    /// `BindingBoundary::pack_tuple(pack_fn, &handles)` when dep[0]
    /// (primary) has DATA in the wave. If only dep[1] (secondary) fires,
    /// settles with RESOLVED (D018 pattern). First-run gate holds until
    /// both deps deliver (R2.5.3 `partial: false`). Post-warmup INVALIDATE
    /// guard: if secondary `prev_data == NO_HANDLE` and batch empty after
    /// warmup, settles with RESOLVED (no stale pair).
    WithLatestFrom { pack_fn: FnId },

    /// `merge(...sources)` — N-dep, forward all DATA handles verbatim
    /// (D022). Zero FFI on fire: no transformation, no binding call.
    /// Each dep's batch handles are retained and emitted individually.
    /// COMPLETE cascades when all deps complete (R1.3.4.b).
    Merge,

    // ----- Slice C-3: flow operators (D024) -----
    /// `take(source, count)` — emits the first `count` DATA values then
    /// self-completes via `Core::complete`. Tracks `count_emitted` in
    /// [`TakeState`](super::op_state::TakeState). When upstream completes
    /// before `count` is reached, the standard auto-cascade propagates
    /// COMPLETE. `count == 0` is allowed: the first fire emits zero
    /// items then immediately self-completes (D027).
    Take { count: u32 },

    /// `skip(source, count)` — drops the first `count` DATA values; once
    /// the threshold is crossed, subsequent DATAs pass through verbatim.
    /// Tracks `count_skipped` in [`SkipState`](super::op_state::SkipState).
    /// On a wave where every input is still in the skip window, queues
    /// DIRTY+RESOLVED to settle (D018 pattern).
    Skip { count: u32 },

    /// `takeWhile(source, predicate)` — emits while `predicate(input)`
    /// holds; on the first `false`, emits any preceding passes then
    /// self-completes via `Core::complete`. Reuses
    /// [`BindingBoundary::predicate_each`] (D029); after the first
    /// `false`, subsequent inputs in the same batch are dropped.
    TakeWhile { fn_id: FnId },

    /// `last(source)` / `last_with_default(source, default)` — buffers
    /// the latest DATA; on upstream COMPLETE, emits `Data(latest)` then
    /// `Complete`. The `default` field is `NO_HANDLE` for the no-default
    /// factory (emits only `Complete` on empty stream), or a registered
    /// default handle (emits `Data(default)` + `Complete` on empty
    /// stream). Storage: [`LastState`](super::op_state::LastState) holds
    /// `latest` (live buffer) and `default` (registration-time, stable).
    /// Opts out of Lock 2.B auto-cascade so it can intercept upstream
    /// COMPLETE.
    Last { default: HandleId },
}

/// Registration options for [`Core::register_operator`].
///
/// `equals` controls operator output dedup (R5.7 — defaults to identity).
/// `partial` controls the R2.5.3 first-run gate (R5.4 — operator dispatch
/// fires on first DATA from any dep when `true`; default `false` matches
/// the gated derived discipline).
#[derive(Copy, Clone, Debug)]
pub struct OperatorOpts {
    pub equals: EqualsMode,
    pub partial: bool,
}

impl Default for OperatorOpts {
    fn default() -> Self {
        Self {
            equals: EqualsMode::Identity,
            partial: false,
        }
    }
}

/// Closure-form fn id OR typed operator discriminant — the two dispatch
/// paths a node can use. State / passthrough nodes pass `None` to
/// [`Core::register`] (no fn at all).
#[derive(Copy, Clone, Debug)]
pub enum NodeFnOrOp {
    /// Closure-form: invokes [`BindingBoundary::invoke_fn`] per fire.
    /// Used for Derived / Dynamic / Producer.
    Fn(FnId),
    /// Typed-op: routes to a `fire_op_*` helper that calls per-operator
    /// FFI methods (`project_each` / `predicate_each` / `fold_each` /
    /// `pairwise_pack` / `pack_tuple`). Used for Operator nodes.
    Op(OperatorOp),
}

/// Pause behavior mode (canonical-spec §2.6 — three modes shipped in TS;
/// Slice F audit, 2026-05-07 — closed the Rust port gap).
///
/// | Mode | Outgoing tier-3 routing while paused | RESUME behavior |
/// |---|---|---|
/// | [`PausableMode::Default`] | suppress fn-fire upstream (no DIRTY emitted) | fire fn ONCE on RESUME if any dep delivered DATA during pause; collapses N pause-window writes into one settle |
/// | [`PausableMode::ResumeAll`] | buffer outgoing tier-3 / tier-4 messages per-wave | replay each buffered wave verbatim on RESUME |
/// | [`PausableMode::Off`] | dispatcher ignores PAUSE; tier-3 flushes immediately | no-op (no buffer to drain) |
///
/// Default is [`PausableMode::Default`] per canonical §2.6 — every untagged
/// source picks it up. Memory profile is O(1) per node (no buffer); the
/// trade-off is "subscribers see one consolidated DATA on RESUME" rather
/// than the K mid-pause emissions verbatim.
///
/// Note: tier-1 (DIRTY) / tier-2 (PAUSE/RESUME) / tier-5 (COMPLETE/ERROR) /
/// tier-6 (TEARDOWN) bypass pause regardless of mode — they remain
/// observable so leaked pause-controllers cannot strand subscribers.
#[derive(Copy, Clone, Debug, Default, PartialEq, Eq)]
pub enum PausableMode {
    /// Suppress fn-fire while paused; fire once on RESUME if any dep
    /// delivered DATA during the pause window. Canonical default.
    #[default]
    Default,
    /// Buffer outgoing tier-3 / tier-4 messages per-wave; replay on
    /// RESUME. Use when subscribers need verbatim emit history (e.g. an
    /// audit log, replay-on-reconnect bridge).
    ResumeAll,
    /// Dispatcher ignores PAUSE for this node — tier-3 flushes
    /// immediately even while a lock is held. Use for nodes whose value
    /// production is intrinsically pause-immune (telemetry counters,
    /// monotonic timers).
    Off,
}

/// Per-kind opts for [`Core::register`]. Cross-kind config knobs live
/// here; per-kind specifics (deps, fn_or_op) live on
/// [`NodeRegistration`].
#[derive(Copy, Clone, Debug)]
pub struct NodeOpts {
    /// Initial cached value. Only valid for state nodes (no deps + no
    /// fn + no op). [`NO_HANDLE`] starts the node sentinel.
    pub initial: HandleId,
    /// Equality mode for outgoing emissions (R1.3.2). Defaults to
    /// [`EqualsMode::Identity`].
    pub equals: EqualsMode,
    /// First-run gate (R2.5.3 / D011). When `true`, the node fires as
    /// soon as ANY dep delivers a real handle; when `false` (default),
    /// the node holds until every dep has delivered.
    pub partial: bool,
    /// Dynamic flag (R2.5.3) — fn declares actually-tracked dep indices
    /// per fire. Only meaningful when `fn_or_op == Some(Fn(_))` AND
    /// deps non-empty.
    pub is_dynamic: bool,
    /// Pause behavior mode (canonical §2.6). Default is
    /// [`PausableMode::Default`]. See [`PausableMode`] for the trade-offs.
    pub pausable: PausableMode,
    /// Replay buffer cap (canonical R2.6.5 / Lock 6.G — Slice E1, 2026-05-07).
    /// `None` (default) disables; `Some(N)` keeps a circular buffer of the
    /// last N DATA emissions and replays them to late subscribers as part
    /// of the per-tier handshake (between [`Message::Start`] and any
    /// terminal slice). Only DATA is buffered; RESOLVED entries are NOT
    /// (R2.6.5 explicit "DATA only").
    pub replay_buffer: Option<usize>,
}

impl Default for NodeOpts {
    fn default() -> Self {
        Self {
            initial: NO_HANDLE,
            equals: EqualsMode::Identity,
            partial: false,
            is_dynamic: false,
            pausable: PausableMode::Default,
            replay_buffer: None,
        }
    }
}

/// Unified node-registration descriptor (D030, Slice D).
///
/// All node kinds (State / Producer / Derived / Dynamic / Operator)
/// register through [`Core::register`] with a `NodeRegistration`. The
/// kind is **derived from the field shape** of the registration —
/// `(deps.is_empty(), fn_or_op variant)`:
///
/// | deps      | fn_or_op   | is_dynamic | resulting kind |
/// |-----------|-----------|-----------|----------------|
/// | empty     | None      | -         | State          |
/// | empty     | Some(Fn)  | -         | Producer       |
/// | non-empty | Some(Fn)  | false     | Derived        |
/// | non-empty | Some(Fn)  | true      | Dynamic        |
/// | non-empty | Some(Op)  | -         | Operator       |
///
/// The sugar wrappers ([`Core::register_state`], [`Core::register_producer`],
/// etc.) build a `NodeRegistration` and delegate.
#[derive(Clone, Debug)]
pub struct NodeRegistration {
    /// Upstream deps in declaration order. Empty for state / producer.
    pub deps: Vec<NodeId>,
    /// Closure-form fn id or typed-op discriminant. `None` for state /
    /// passthrough.
    pub fn_or_op: Option<NodeFnOrOp>,
    /// Cross-kind config knobs.
    pub opts: NodeOpts,
}

/// Equality mode for a node's outgoing emissions.
///
/// `Identity` is the default: cache vs. new handle compare is a `u64` equal —
/// zero FFI. `Custom` invokes [`BindingBoundary::custom_equals`] every check
/// (R1.3.2.b two-arg call when both sides are non-sentinel).
#[derive(Copy, Clone, Debug)]
pub enum EqualsMode {
    Identity,
    Custom(FnId),
}

/// Internal identifier for a single subscription. Allocated per
/// [`Core::subscribe`] call. Wrapped by [`Subscription`] for the public API;
/// consumed directly only by Core internals and the [`Subscription::Drop`]
/// path.
#[derive(Copy, Clone, Eq, PartialEq, Hash, Debug)]
pub(crate) struct SubscriptionId(u64);

/// RAII subscription handle.
///
/// Returned by [`Core::subscribe`]. While the handle is held, the sink stays
/// registered against its node. Dropping the handle (explicitly via
/// `drop(sub)` or implicitly at scope exit) unsubscribes the sink — no manual
/// `unsubscribe()` call is needed. Per §10.12 of the rust-port session doc.
///
/// # Lifetime semantics
///
/// The subscription holds a [`Weak`] reference back to the Core's state. If
/// the Core is dropped before the subscription, the Drop impl is a silent
/// no-op (the sink has nowhere to deregister from anyway). This avoids a
/// reference cycle when subscribers capture an `Arc<Core>` in their closure.
///
/// # Thread safety
///
/// `Send + Sync`. The handle can be moved across threads or dropped from
/// any thread.
///
/// # Not Clone
///
/// `Subscription` owns the unsubscribe action exclusively. Cloning would
/// require either "first drop wins" or "last drop wins" semantics, both
/// of which surprise. If a binding needs multiple deregistration handles,
/// it should subscribe multiple times (each producing a fresh handle) or
/// wrap the single `Subscription` in `Arc<Mutex<Option<Subscription>>>`.
#[must_use = "dropping a Subscription unsubscribes its sink immediately"]
pub struct Subscription {
    state: Weak<Mutex<CoreState>>,
    node_id: NodeId,
    sub_id: SubscriptionId,
}

impl Subscription {
    /// The node this subscription is attached to.
    #[must_use]
    pub fn node_id(&self) -> NodeId {
        self.node_id
    }
}

impl Drop for Subscription {
    fn drop(&mut self) {
        // Silent no-op if Core is gone. This keeps Drop infallible (no panics
        // from a dropped subscription racing a dropped Core) and avoids
        // surprising users with errors on shutdown.
        //
        // Producer deactivation (Slice D, D031): if removing this sub
        // empties the subscribers map AND the node is a producer, fire
        // `BindingBoundary::producer_deactivate(node_id)` AFTER releasing
        // the state lock. The binding then drops its per-node state
        // (subscriptions to upstream sources, captured closure state),
        // which transitively unsubs from upstreams via their own
        // `Subscription::Drop`. Re-entrance into Core from the deactivate
        // hook is permitted since the lock is released first.
        let Some(state) = self.state.upgrade() else {
            return;
        };
        // Slice E2 (D056): when the last subscriber drops, fire the
        // node's OnDeactivation cleanup hook BEFORE producer_deactivate
        // (cleanup may release handles the producer subscription owns;
        // reverse order would let producer_deactivate drop subs that user
        // cleanup expected to be live). Both calls are lock-released per
        // D045.
        //
        // OnDeactivation gating (D068, QA Q3 fix): fires only when the
        // node has fired its fn at least once AND has a fn (`fn_id`
        // populated). State nodes have no fn — they cannot register a
        // cleanup spec via the production fn-return path (R2.4.5), so
        // firing `cleanup_for` on them is wasted FFI; the binding's
        // lookup is guaranteed to find no `current_cleanup`. Skipping
        // here saves the FFI hop and matches the design-doc wording
        // ("never-fired state nodes" — state-with-initial-value satisfies
        // `has_fired_once = true` but still has no fn).
        //
        // Slice E2 /qa Q2(b) (D069): if the node is a resubscribable
        // node that's ALREADY terminal (terminate fired BEFORE this last
        // sub drop), fire `wipe_ctx` lock-released AFTER OnDeactivation
        // + producer_deactivate. Mutually exclusive with `terminate_node`'s
        // queue-wipe site: terminate-with-empty-subs goes through
        // `pending_wipes`; terminate-with-live-subs routes here when
        // those subs eventually drop. Either path fires exactly one
        // wipe per terminal lifecycle.
        let (was_last_sub, is_producer, has_user_cleanup, fire_wipe, binding) = {
            let mut s = state.lock();
            let Some(rec) = s.nodes.get_mut(&self.node_id) else {
                return;
            };
            rec.subscribers.remove(&self.sub_id);
            // Slice X4 / D2: bump revision so any pending_notify entry for
            // this node opened earlier in the wave starts a fresh batch on
            // the next queue_notify, dropping the now-departed sink from
            // the snapshot.
            rec.subscribers_revision = rec.subscribers_revision.wrapping_add(1);
            let last = rec.subscribers.is_empty();
            let producer = rec.is_producer();
            // OnDeactivation gate: must have run a fn at least once
            // (has_fired_once) AND have a fn registered (fn_id.is_some()).
            // The fn_id check excludes state nodes whose has_fired_once
            // tracks initial-value status, not "user fn ran."
            let user_cleanup = rec.has_fired_once && rec.fn_id.is_some();
            let fire_wipe = last && rec.resubscribable && rec.terminal.is_some();
            // Clone the binding Arc out only if at least one hook will
            // fire. Cheap (Arc::clone) in the common path; skipped on
            // non-last-sub or never-fired non-producer nodes.
            let binding = if last && (producer || user_cleanup || fire_wipe) {
                Some(s.binding.clone())
            } else {
                None
            };
            (last, producer, user_cleanup, fire_wipe, binding)
        };
        if was_last_sub {
            if let Some(binding) = binding {
                if has_user_cleanup {
                    binding.cleanup_for(self.node_id, CleanupTrigger::OnDeactivation);
                }
                if is_producer {
                    binding.producer_deactivate(self.node_id);
                }
                // D069: eager wipe — fires AFTER OnDeactivation so the
                // user closure observes pre-wipe `store` (matches the
                // existing "OnDeactivation runs before wipe on terminal
                // reset" invariant covered by test 10). Idempotent —
                // `HashMap::remove` on absent key is a no-op, so even
                // if the wave already drained `pending_wipes` earlier,
                // this fire is benign.
                if fire_wipe {
                    binding.wipe_ctx(self.node_id);
                }
            }
        }
    }
}

// Compile-time assertion that Subscription is Send + Sync. If a future field
// breaks this, the build fails here rather than downstream at the binding
// site.
const _: fn() = || {
    fn assert_send_sync<T: Send + Sync>() {}
    assert_send_sync::<Subscription>();
};

/// A subscriber callback. `Send + Sync` so the Core can fire it from any
/// thread; `Fn` (not `FnMut`) so multiple references coexist — capture
/// mutable state in `Mutex<T>` or atomics on the binding side.
pub type Sink = Arc<dyn Fn(&[Message]) + Send + Sync>;

// ---------------------------------------------------------------------------
// PAUSE/RESUME state — §10.2 of the rust-port session doc
// ---------------------------------------------------------------------------

/// Per-node pause state.
///
/// Replaces the four TS fields (`_pauseLocks`, `_pauseBuffer`,
/// `_pauseDroppedCount`, `_pauseStartNs`) with a single enum where
/// the buffered fields are unreachable in the [`Self::Active`] variant —
/// the compiler refuses access. Per §10.2 simplification.
///
/// # Invariants
///
/// - `Active` ⇔ no lockId held.
/// - `Paused { locks, .. }` ⇔ `!locks.is_empty()`.
/// - Buffered messages are tier 3 (DATA/RESOLVED) and tier 4 (INVALIDATE)
///   only. Other tiers pass through immediately even while paused.
/// - `dropped` counts messages that fell out the front of `buffer` due to
///   the Core-global `pause_buffer_cap`; it is reported on resume so callers
///   can detect overflow without re-tracking it externally.
#[derive(Debug)]
pub(crate) enum PauseState {
    Active,
    Paused {
        /// Active lock holders. `SmallVec` keeps the common 1–2 lock case
        /// stack-allocated. Replaces `Set<unknown>` from TS.
        locks: SmallVec<[LockId; 2]>,
        /// Buffered tier-3/tier-4 outgoing messages, in arrival order.
        /// Replayed on the final RESUME.
        buffer: VecDeque<Message>,
        /// Count of messages dropped from the front when `buffer.len()` would
        /// exceed `pause_buffer_cap`. Cleared on final RESUME (next pause
        /// cycle starts fresh).
        dropped: u32,
        /// Wall-clock-monotonic ns when the lock first transitioned this node
        /// from `Active` to `Paused`. Used by R1.3.8.c overflow ERROR
        /// synthesis to compute `lock_held_duration_ms` in the diagnostic
        /// payload (Slice F, A3 — 2026-05-07).
        started_at_ns: u64,
        /// True after the first overflow event in this pause cycle has been
        /// reported via [`crate::boundary::BindingBoundary::synthesize_pause_overflow_error`].
        /// Subsequent overflows in the same cycle don't re-emit ERROR
        /// (canonical R1.3.8.c: "once per overflow event"). Cleared on
        /// final RESUME (next pause cycle starts fresh).
        overflow_reported: bool,
        /// Default-mode bookkeeping (Slice F audit close, 2026-05-07).
        /// Set to `true` when an upstream dep delivery arrives while this
        /// node is paused with [`PausableMode::Default`]. On final RESUME,
        /// if `true`, the node is added back to `pending_fires` so the fn
        /// fires once with the consolidated dep state. Always `false` for
        /// `ResumeAll` mode (the buffered messages are the consolidation
        /// mechanism there). Cleared on final RESUME.
        pending_wave: bool,
    },
}

impl PauseState {
    pub(crate) fn is_paused(&self) -> bool {
        matches!(self, Self::Paused { .. })
    }

    fn lock_count(&self) -> usize {
        match self {
            Self::Active => 0,
            Self::Paused { locks, .. } => locks.len(),
        }
    }

    fn contains_lock(&self, lock_id: LockId) -> bool {
        match self {
            Self::Active => false,
            Self::Paused { locks, .. } => locks.contains(&lock_id),
        }
    }

    /// Add a lock; transitions Active → Paused on first lock. Idempotent on
    /// duplicate lock_id (matches TS convention; spec is silent on the case).
    fn add_lock(&mut self, lock_id: LockId) {
        match self {
            Self::Active => {
                let mut locks = SmallVec::new();
                locks.push(lock_id);
                *self = Self::Paused {
                    locks,
                    buffer: VecDeque::new(),
                    dropped: 0,
                    started_at_ns: monotonic_ns(),
                    overflow_reported: false,
                    pending_wave: false,
                };
            }
            Self::Paused { locks, .. } => {
                if !locks.contains(&lock_id) {
                    locks.push(lock_id);
                }
            }
        }
    }

    /// Mark that an upstream dep delivered DATA to a node paused with
    /// [`PausableMode::Default`]. The node will re-enter `pending_fires`
    /// on final RESUME via [`Self::take_pending_wave`].
    pub(crate) fn mark_pending_wave(&mut self) {
        if let Self::Paused { pending_wave, .. } = self {
            *pending_wave = true;
        }
    }

    /// Read and clear the `pending_wave` flag. Called from
    /// [`Core::resume`] when transitioning Paused → Active. Returns `true`
    /// only if the node was paused with `pending_wave` set.
    pub(crate) fn take_pending_wave(&mut self) -> bool {
        if let Self::Paused { pending_wave, .. } = self {
            std::mem::replace(pending_wave, false)
        } else {
            false
        }
    }

    /// Remove a lock; if the lockset becomes empty, transition Paused →
    /// Active and return the buffered messages for replay (along with the
    /// dropped count for diagnostics). Unknown lock_id is an idempotent
    /// no-op (matches TS, R1.2.6 implicit).
    fn remove_lock(&mut self, lock_id: LockId) -> Option<(VecDeque<Message>, u32)> {
        match self {
            Self::Active => None,
            Self::Paused { locks, .. } => {
                if let Some(idx) = locks.iter().position(|l| *l == lock_id) {
                    locks.swap_remove(idx);
                }
                if locks.is_empty() {
                    let prev = std::mem::replace(self, Self::Active);
                    if let Self::Paused {
                        buffer, dropped, ..
                    } = prev
                    {
                        return Some((buffer, dropped));
                    }
                }
                None
            }
        }
    }

    /// Append a message to the buffer; if the buffer would exceed `cap`,
    /// pop from the front (oldest-first), increment `dropped`, and return
    /// the dropped messages so the caller can release any payload handles
    /// they reference. `cap` of `None` means unbounded.
    ///
    /// Returns [`PushBufferedResult`] carrying both the dropped messages
    /// (for refcount release) and whether this push triggered the FIRST
    /// overflow event in the current pause cycle (for R1.3.8.c ERROR
    /// synthesis — the caller schedules a single ERROR per cycle).
    ///
    /// Note: refcount management for the message's payload handle is the
    /// caller's responsibility — see [`Core::queue_notify`] for the
    /// retain/release discipline. The buffer itself is just a message
    /// container; refcounts cross the binding boundary.
    pub(crate) fn push_buffered(&mut self, msg: Message, cap: Option<usize>) -> PushBufferedResult {
        let mut result = PushBufferedResult::default();
        if let Self::Paused {
            buffer,
            dropped,
            overflow_reported,
            ..
        } = self
        {
            buffer.push_back(msg);
            if let Some(c) = cap {
                while buffer.len() > c {
                    if let Some(dropped_msg) = buffer.pop_front() {
                        result.dropped_msgs.push(dropped_msg);
                    }
                    *dropped = dropped.saturating_add(1);
                }
            }
            // R1.3.8.c (Slice F, A3): flag first overflow this cycle.
            if !result.dropped_msgs.is_empty() && !*overflow_reported {
                *overflow_reported = true;
                result.first_overflow_this_cycle = true;
            }
        }
        result
    }

    /// Snapshot the diagnostic for an R1.3.8.c overflow ERROR synthesis.
    /// Returns `(dropped_count, lock_held_ns)`. Caller must already know
    /// the configured cap (it's a Core-global value, not per-PauseState).
    pub(crate) fn overflow_diagnostic(&self) -> Option<(u32, u64)> {
        match self {
            Self::Active => None,
            Self::Paused {
                dropped,
                started_at_ns,
                ..
            } => {
                let lock_held_ns = monotonic_ns().saturating_sub(*started_at_ns);
                Some((*dropped, lock_held_ns))
            }
        }
    }
}

/// Return shape for [`PauseState::push_buffered`]. Carries both the dropped
/// messages (for refcount release) and an "is this the first overflow this
/// cycle" flag (for R1.3.8.c ERROR synthesis scheduling).
#[derive(Default)]
pub(crate) struct PushBufferedResult {
    pub(crate) dropped_msgs: Vec<Message>,
    pub(crate) first_overflow_this_cycle: bool,
}

/// Pending R1.3.8.c overflow ERROR synthesis entry. Recorded by
/// [`Core::queue_notify`] when the pause buffer first overflows in a cycle;
/// drained at wave-end after the lock-released call to
/// [`crate::boundary::BindingBoundary::synthesize_pause_overflow_error`].
///
/// `configured_max` is captured at scheduling time rather than read at
/// drain — the user could change `pause_buffer_cap` between schedule and
/// drain, and the diagnostic reads "the cap that was in effect when the
/// overflow happened."
#[derive(Debug, Clone)]
pub(crate) struct PendingPauseOverflow {
    pub(crate) node_id: NodeId,
    pub(crate) dropped_count: u32,
    pub(crate) configured_max: usize,
    pub(crate) lock_held_ns: u64,
}

/// Errors returnable by [`Core::pause`] and [`Core::resume`].
#[derive(Error, Debug, Clone, PartialEq)]
pub enum PauseError {
    #[error("pause/resume: unknown node {0:?}")]
    UnknownNode(NodeId),
}

/// Errors returnable by [`Core::up`] (canonical R1.4.1).
#[derive(Error, Debug, Clone, PartialEq)]
pub enum UpError {
    /// Node id is not registered.
    #[error("up: unknown node {0:?}")]
    UnknownNode(NodeId),
    /// Tier-3 (DATA / RESOLVED) and tier-5 (COMPLETE / ERROR) are
    /// downstream-only per R1.4.1; rejected at the boundary.
    #[error(
        "up: tier {tier} is forbidden upstream — value (tier 3) and \
         terminal-lifecycle (tier 5) planes are downstream-only per R1.4.1"
    )]
    TierForbidden { tier: u8 },
}

/// Errors returnable by [`Core::register`] and its sugar wrappers
/// ([`Core::register_state`], [`Core::register_producer`],
/// [`Core::register_derived`], [`Core::register_dynamic`],
/// [`Core::register_operator`]).
///
/// Slice H (2026-05-07) promoted these from `assert!`/`panic!` to typed
/// errors so that callers can recover from contract violations without
/// process abort. Every variant corresponds to a construction-time
/// invariant that the caller is responsible for upholding; the dispatcher
/// rejects the registration before any reactive state is created (so
/// there is no `Message::Error` channel through which to surface the
/// failure — these are imperative-layer errors, not reactive ones).
///
/// All variants are zero-side-effect: when [`Core::register`] returns
/// `Err`, no node has been added to the graph and any handle retains
/// taken on the way in (e.g. operator scratch seed retains via
/// [`BindingBoundary::retain_handle`]) have been released.
#[derive(Error, Debug, Clone, PartialEq, Eq)]
pub enum RegisterError {
    /// One of the supplied dep ids is not a registered node.
    #[error("register: unknown dep {0:?}")]
    UnknownDep(NodeId),

    /// `op` was supplied (operator node) but `deps` was empty. Operator
    /// nodes need at least one dep — for subscription-managed combinators
    /// with no declared deps, use [`Core::register_producer`] instead.
    #[error(
        "register: operator nodes require at least one dep — \
         use register_producer for subscription-managed combinators"
    )]
    OperatorWithoutDeps,

    /// [`NodeOpts::initial`] was set to a real handle but the registration
    /// shape is not a state node (state nodes are `deps.is_empty() &&
    /// fn_id.is_none() && op.is_none()`). Initial cache only makes sense
    /// for state nodes.
    #[error("register: NodeOpts::initial only valid for state nodes (no deps + no fn + no op)")]
    InitialOnlyForStateNodes,

    /// A supplied dep is terminal (COMPLETE / ERROR) AND not
    /// resubscribable. Adding it would create a permanent wedge — the dep
    /// will never re-emit, so the registered node would be stuck.
    /// Mirrors [`SetDepsError::TerminalDep`] at registration time.
    #[error(
        "register: dep {0:?} is terminal and not resubscribable; \
         mark it resubscribable before terminating, or remove it from the dep list"
    )]
    TerminalDep(NodeId),

    /// A stateful operator ([`OperatorOp::Scan`] / [`OperatorOp::Reduce`])
    /// was registered with `seed = NO_HANDLE`. R2.5.3 first-run gate
    /// requires the seed to be a real handle so that the operator can
    /// emit on its first fire.
    #[error("register: operator seed must be a real handle (R2.5.3); got NO_HANDLE")]
    OperatorSeedSentinel,
}

/// Errors returnable by [`Core::set_pausable_mode`].
///
/// Slice H (2026-05-07) promoted these from `assert!`/`panic!` to typed
/// errors. Same imperative-layer error model as [`RegisterError`].
#[derive(Error, Debug, Clone, PartialEq, Eq)]
pub enum SetPausableModeError {
    /// `node_id` is not a registered node.
    #[error("set_pausable_mode: unknown node {0:?}")]
    UnknownNode(NodeId),
    /// The node currently holds at least one pause lock. Changing pausable
    /// mode mid-pause would lose buffered content or strand a
    /// `pending_wave` flag — resume all locks first.
    #[error(
        "set_pausable_mode: cannot change pausable mode while paused; \
         resume all locks first"
    )]
    WhilePaused,
}

/// Per-dep record. Replaces the parallel `deps` / `dep_handles` /
/// `dep_terminals` vectors from v1. Canonical spec R2.9.b alignment.
///
/// Each entry tracks one dep's lifecycle state, wave-scoped batch data,
/// and cross-wave `prev_data` for `ctx.prevData` access.
pub(crate) struct DepRecord {
    /// The dep node this record tracks.
    pub(crate) node: NodeId,
    /// Last DATA handle from the end of the previous wave. [`NO_HANDLE`]
    /// means the dep has never emitted DATA.
    pub(crate) prev_data: HandleId,
    /// Per-dep dirty flag — awaiting DATA/RESOLVED for current wave.
    pub(crate) dirty: bool,
    /// Per-dep involved-this-wave flag. Distinguishes:
    /// - `involved && data_batch.is_empty()` → dep settled RESOLVED
    /// - `!involved && data_batch.is_empty()` → dep was not in this wave
    pub(crate) involved_this_wave: bool,
    /// DATA handles accumulated this wave. Outside `batch()` scope, at most
    /// 1 element. Inside `batch()`, K emits on the source produce K entries
    /// per R1.3.6.b coalescing. Each handle holds a `retain_handle` share
    /// taken at `deliver_data_to_consumer` time; released at wave-end
    /// rotation in `clear_wave_state`.
    pub(crate) data_batch: SmallVec<[HandleId; 1]>,
    /// Terminal state for this dep. `None` = dep is live.
    /// `Some` = dep emitted COMPLETE/ERROR. When ALL entries are Some,
    /// the node auto-cascades per Lock 2.B (ERROR dominates COMPLETE).
    pub(crate) terminal: Option<TerminalKind>,
}

impl DepRecord {
    fn new(node: NodeId) -> Self {
        Self {
            node,
            prev_data: NO_HANDLE,
            dirty: false,
            involved_this_wave: false,
            data_batch: SmallVec::new(),
            terminal: None,
        }
    }
}

/// Internal node record. Mirrors `core.ts:132–154` post-D030 unification.
///
/// **Kind is derived, not stored** (D030, Slice D). `(dep_records.is_empty(),
/// fn_id, op, is_dynamic)` uniquely identifies the kind — see [`NodeKind`].
/// Helper methods (`is_state()`, `is_producer()`, `is_compute()`,
/// `is_operator()`, `skips_auto_cascade()`, `kind()`) cover the common
/// predicates without unpacking via [`Core::kind_of`].
///
/// The 5 bool fields (`has_fired_once`, `dirty`, `involved_this_wave`,
/// `has_received_teardown`, `resubscribable`, `is_dynamic`) each represent
/// an orthogonal concern. `is_dynamic` is constant per node (set at
/// register time); the others are mutable lifecycle state. Collapsing
/// them into a bitfield would obscure intent.
#[allow(clippy::struct_excessive_bools)]
pub(crate) struct NodeRecord {
    /// Per-dep records. Replaces the old parallel `deps` / `dep_handles` /
    /// `dep_terminals` vecs. Dep NodeIds derived via `dep_ids()`.
    pub(crate) dep_records: Vec<DepRecord>,
    /// User-fn id for closure-form dispatch. `Some` for Derived / Dynamic /
    /// Producer; `None` for State / Operator. (Operator dispatch goes via
    /// [`Self::op`] instead.)
    pub(crate) fn_id: Option<FnId>,
    /// Operator discriminant for typed-op dispatch. `Some` for Operator
    /// nodes; `None` otherwise. Mutually exclusive with `fn_id` (a node is
    /// either closure-form OR typed-op, never both).
    pub(crate) op: Option<OperatorOp>,
    /// True for Dynamic nodes (R2.5.3 — fn declares actually-tracked dep
    /// indices per fire). False for everything else. Only meaningful when
    /// `fn_id.is_some()` AND `!dep_records.is_empty()`.
    pub(crate) is_dynamic: bool,
    pub(crate) equals: EqualsMode,

    // Mutable state
    pub(crate) cache: HandleId,
    pub(crate) has_fired_once: bool,
    pub(crate) subscribers: HashMap<SubscriptionId, Sink>,
    /// Monotonic counter bumped on every mutation of [`Self::subscribers`]
    /// (insert on subscribe, remove on `Subscription::Drop`, remove on
    /// handshake-panic cleanup). Used by
    /// [`crate::batch::Core::queue_notify`] to detect mid-wave subscriber-
    /// set changes and start a fresh `PendingBatch` with an updated sink
    /// snapshot — closes D2 (Slice X4, 2026-05-08): the late-subscriber
    /// and multi-emit-per-wave gap where the pre-fix per-node single
    /// snapshot meant a sub installed between two emits to the same node
    /// in one wave was invisible to the second emit's flush.
    ///
    /// Per-node (not per-Core) so that a subscribe to node A doesn't
    /// invalidate snapshot reuse for node B's pending batch in the same
    /// wave.
    pub(crate) subscribers_revision: u64,
    /// For dynamic nodes: which dep indices fn actually tracks.
    /// For static derived: all indices, populated at construction.
    pub(crate) tracked: HashSet<usize>,

    // Wave-scoped state — cleared at wave end.
    pub(crate) dirty: bool,
    pub(crate) involved_this_wave: bool,

    /// Per-node pause state. Default `Active`. See [`PauseState`].
    pub(crate) pause_state: PauseState,
    /// Pause behavior mode (canonical-spec §2.6). Set at registration via
    /// [`NodeOpts::pausable`]. Default [`PausableMode::Default`] suppresses
    /// fn-fire while paused and consolidates N pause-window dep deliveries
    /// into one fn-fire on RESUME; `ResumeAll` buffers tier-3/4 outgoing
    /// for verbatim replay; `Off` ignores PAUSE entirely. See
    /// [`PausableMode`].
    pub(crate) pausable: PausableMode,
    /// Replay buffer cap (R2.6.5 / Lock 6.G — Slice E1, 2026-05-07).
    /// `None` disables; `Some(N)` keeps a circular VecDeque of the last N
    /// DATA-handle emissions for late-subscriber replay. Each handle in
    /// the buffer owns one binding-side retain share, released on evict
    /// (cap exceeded) or in `Drop for CoreState`.
    pub(crate) replay_buffer_cap: Option<usize>,
    pub(crate) replay_buffer: VecDeque<HandleId>,

    /// Terminal lifecycle state for THIS node's outgoing stream. Once set,
    /// further `emit` calls are silent no-ops, fn no longer fires, and only
    /// the terminal message has been queued downstream.
    pub(crate) terminal: Option<TerminalKind>,
    /// True after the first TEARDOWN has been processed for this node
    /// (R2.6.4 / Lock 6.F). Subsequent TEARDOWN deliveries are idempotent
    /// — the auto-prepended COMPLETE only fires on the first one. Without
    /// this flag, a redundant TEARDOWN delivered via the cascade plus an
    /// explicit `core.teardown(node)` would re-emit `[COMPLETE, TEARDOWN]`
    /// to subscribers per delivery, which is incorrect.
    pub(crate) has_received_teardown: bool,
    /// Per R2.2.7 / R2.5.3 — resubscribable terminal lifecycle.
    /// When `true` AND `terminal == Some(...)`, a fresh subscribe call
    /// will reset the node: clear `terminal`, `has_fired_once`,
    /// `has_received_teardown`, all dep_records to sentinel, and drain the
    /// pause lockset. Default `false`.
    pub(crate) resubscribable: bool,
    /// Meta companion nodes attached to this node per R1.3.9.d. When this
    /// node tears down, its meta companions are torn down FIRST (before
    /// the main node's auto-COMPLETE + TEARDOWN wire emission), so
    /// observers see companions terminate before the parent. The ordering
    /// is load-bearing — meta nodes typically subscribe to parent state
    /// that becomes inconsistent during the parent's destruction phase.
    pub(crate) meta_companions: Vec<NodeId>,
    /// R5.4 / D011 partial-mode: when `true`, fire_fn skips the R2.5.3
    /// first-run gate — the node fires as soon as ANY dep delivers a
    /// real handle, even if other deps remain sentinel. Defaults to
    /// `false` (gated). Lifted into Core for operator support; for
    /// State/Derived/Dynamic nodes the field is settable but the gated
    /// path remains the typical caller default.
    pub(crate) partial: bool,
    /// Generic per-operator scratch slot (Slice C-3, D026). Replaces
    /// the typed `operator_state: HandleId` field used by Slices C-1 / C-2.
    /// `None` for non-operator kinds and operators with no cross-wave
    /// state (Map / Filter / Combine / WithLatestFrom / Merge); `Some`
    /// for stateful operators ([`OperatorOp::Scan`] / [`Reduce`] /
    /// [`DistinctUntilChanged`] / [`Pairwise`] / [`Take`] / [`Skip`] /
    /// [`TakeWhile`] / [`Last`]).
    ///
    /// The boxed value implements
    /// [`OperatorScratch`](crate::op_state::OperatorScratch); its
    /// `release_handles` method is called from
    /// [`reset_for_fresh_lifecycle`] (resubscribable terminal cycle) and
    /// from [`Drop for CoreState`].
    ///
    /// **Refcount discipline:** the state struct owns whatever handle
    /// shares it stores (e.g., [`ScanState::acc`](crate::op_state::ScanState::acc),
    /// [`LastState::latest`](crate::op_state::LastState::latest)).
    /// Per-fire helpers retain the new value before releasing the old;
    /// `release_handles` releases the current shares at end-of-life.
    pub(crate) op_scratch: Option<Box<dyn crate::op_state::OperatorScratch>>,
}

impl NodeRecord {
    // ---- Kind predicates (D030 — derived from field shape) ----

    /// True iff this is a state node (no deps, no fn, no op).
    pub(crate) fn is_state(&self) -> bool {
        self.dep_records.is_empty() && self.fn_id.is_none() && self.op.is_none()
    }

    /// True iff this is a producer node (no deps + has fn + no op).
    /// Producers fire fn once on first subscribe; cleanup fires via
    /// [`BindingBoundary::producer_deactivate`] (D031, Slice D).
    pub(crate) fn is_producer(&self) -> bool {
        self.dep_records.is_empty() && self.fn_id.is_some() && self.op.is_none()
    }

    /// True iff this is a compute node (Derived / Dynamic / Operator) —
    /// has at least one dep AND either a fn or an op.
    #[allow(dead_code)] // Convenience predicate; callers may use is_state/is_producer instead.
    pub(crate) fn is_compute(&self) -> bool {
        !self.dep_records.is_empty() && (self.fn_id.is_some() || self.op.is_some())
    }

    /// True iff this is an Operator node (has op set).
    #[allow(dead_code)] // Direct `op.is_some()` is more common; this is a readability sugar.
    pub(crate) fn is_operator(&self) -> bool {
        self.op.is_some()
    }

    /// True iff this node opts OUT of Lock 2.B auto-cascade —
    /// Operator(Reduce) / Operator(Last) intercept upstream COMPLETE.
    pub(crate) fn skips_auto_cascade(&self) -> bool {
        match self.op {
            Some(op) => NodeKind::Operator(op).skips_auto_cascade(),
            None => false,
        }
    }

    /// Compute the public-API [`NodeKind`] from the field shape (D030).
    /// Used by [`Core::kind_of`] and rare internal sites that need the
    /// enum (most use the predicate methods above).
    pub(crate) fn kind(&self) -> NodeKind {
        if let Some(op) = self.op {
            NodeKind::Operator(op)
        } else if self.dep_records.is_empty() {
            if self.fn_id.is_some() {
                NodeKind::Producer
            } else {
                NodeKind::State
            }
        } else if self.is_dynamic {
            NodeKind::Dynamic
        } else {
            NodeKind::Derived
        }
    }

    // ---- Existing accessors ----

    /// Iterator over dep NodeIds in declaration order.
    pub(crate) fn dep_ids(&self) -> impl Iterator<Item = NodeId> + '_ {
        self.dep_records.iter().map(|r| r.node)
    }

    /// Collected dep NodeIds — for call sites that need a `Vec<NodeId>`.
    pub(crate) fn dep_ids_vec(&self) -> Vec<NodeId> {
        self.dep_ids().collect()
    }

    /// Number of deps.
    pub(crate) fn dep_count(&self) -> usize {
        self.dep_records.len()
    }

    /// True if any dep is in sentinel state (never emitted DATA and no
    /// data this wave). Replaces the old `dep_handles.contains(&NO_HANDLE)`.
    pub(crate) fn has_sentinel_deps(&self) -> bool {
        self.dep_records
            .iter()
            .any(|r| r.prev_data == NO_HANDLE && r.data_batch.is_empty())
    }

    /// Find the index of a dep by NodeId.
    pub(crate) fn dep_index_of(&self, dep_id: NodeId) -> Option<usize> {
        self.dep_records.iter().position(|r| r.node == dep_id)
    }

    /// True if ALL dep terminal slots are populated (Lock 2.B cascade check).
    pub(crate) fn all_deps_terminal(&self) -> bool {
        !self.dep_records.is_empty() && self.dep_records.iter().all(|r| r.terminal.is_some())
    }
}

/// All mutable Core state, behind one [`parking_lot::Mutex`].
///
/// v1 single-mutex; per-subgraph `ReentrantMutex` parallelism is a later
/// optimization (CLAUDE.md Rust invariant 3).
pub(crate) struct CoreState {
    pub(crate) next_node_id: u64,
    pub(crate) next_subscription_id: u64,
    pub(crate) next_lock_id: u64,
    pub(crate) nodes: HashMap<NodeId, NodeRecord>,
    /// Inverted adjacency: `parent → children`. Updated on registration.
    pub(crate) children: HashMap<NodeId, HashSet<NodeId>>,
    /// Nodes whose fn we owe a fire to — drained by [`Core::run_wave`].
    pub(crate) pending_fires: HashSet<NodeId>,
    /// Per-node outgoing message buffer; flushed at wave end. Insertion-
    /// ordered so flush order is deterministic — load-bearing for
    /// R1.3.9.d meta-TEARDOWN ordering: when a parent and its meta
    /// companion both have queued messages in the same wave, the meta
    /// (queued first via `teardown_inner`'s recursion order) flushes
    /// first.
    ///
    /// Each entry carries the per-wave subscriber snapshot taken at first
    /// touch (Slice A close, M1: lock-released drain). Late subscribers
    /// installed mid-wave between fn-fire iterations don't appear in
    /// already-snapshotted entries; this is the load-bearing fix that
    /// prevents duplicate-Data delivery when a handshake delivers the
    /// post-commit cache and the wave's flush would otherwise also fire
    /// to the same sink.
    pub(crate) pending_notify: IndexMap<NodeId, PendingPerNode>,
    pub(crate) in_tick: bool,
    /// Core-global cap on per-node pause replay buffer length. `None` means
    /// unbounded. Per the user direction (Q1, 2026-05-05): start core-global;
    /// per-node override can be added later as a pure addition without API
    /// breakage. Default `None`.
    pub(crate) pause_buffer_cap: Option<usize>,
    /// Core-global cap on wave-drain iterations before
    /// [`crate::batch::Core::drain_and_flush`] aborts with a diagnostic panic.
    /// Replaces the prior `MAX_DRAIN_ITERATIONS` hard-coded constant
    /// (R4.3 / Lock 2.F′). Default `10_000`.
    ///
    /// The drain loop bound exists to surface runtime cycles
    /// (e.g. an operator that re-arms its own `pending_fires` slot during
    /// `invoke_fn`) as a panic with context, rather than letting Core
    /// spin forever. Structural cycles via [`Core::set_deps`] are
    /// rejected at edge-mutation time (`SetDepsError::WouldCreateCycle`);
    /// registration is structurally cycle-safe by construction (the new
    /// node's id is not allocated until AFTER deps are validated, so deps
    /// cannot transitively reach the new node). The drain bound is the
    /// safety net for runtime cycles that bypass both static checks.
    pub(crate) max_batch_drain_iterations: u32,
    /// Deferred sink-fire jobs collected by `flush_notifications`. The wave
    /// engine populates this under the state lock during the flush phase;
    /// `run_wave` then drops the lock and fires the jobs. Each tuple is
    /// `(sinks_for_one_node_one_phase, phase_messages)`. Empty between waves.
    pub(crate) deferred_flush_jobs: crate::batch::DeferredJobs,
    /// Payload-handle releases owed for messages that landed in
    /// `pending_notify` during this wave (one per `payload_handle()`).
    /// `run_wave` releases these after sinks fire and the lock is dropped,
    /// balancing the retain done in `queue_notify`.
    pub(crate) deferred_handle_releases: Vec<HandleId>,
    /// Binding-boundary handle for `Drop`-time refcount balancing.
    /// `Core` also holds a clone of this Arc; storing it here lets
    /// `Drop for CoreState` walk every retained slot and release the
    /// binding-side share when the last `Core` clone drops. Without this,
    /// `cache` / `terminal` / `dep_terminals` Error / pause-buffer payload
    /// handle refs leak in the binding registry until process exit.
    pub(crate) binding: Arc<dyn BindingBoundary>,
    /// Pre-wave cache snapshots used to restore state if the wave aborts
    /// mid-flight (e.g., a `Core::batch` closure panics). Each entry is
    /// `(node_id → old_cache_handle)` — the handle the node held BEFORE
    /// the wave started writing to it. The snapshotted handle holds a
    /// retain (taken when the snapshot was inserted) so it stays alive
    /// for restoration. On wave success, snapshots are dropped and their
    /// retains released. On wave abort (`BatchGuard::drop` panic-discard
    /// path), each cache slot is restored from the snapshot — the slot's
    /// current handle is released, and the snapshot's retain transfers
    /// to the cache slot. Only populated for in-flight waves; empty
    /// between waves.
    pub(crate) wave_cache_snapshots: HashMap<NodeId, HandleId>,
    /// Nodes that need an auto-Resolved at wave end if they don't receive
    /// a tier-3+ message from their own commit_emission. Populated by
    /// the RESOLVED child propagation in `commit_emission` (which queues
    /// Dirty but defers Resolved to avoid double-settlement). Drained by
    /// the auto-resolve sweep in `drain_and_flush`. Cleared by
    /// `clear_wave_state`.
    pub(crate) pending_auto_resolve: ahash::AHashSet<NodeId>,
    /// Topology-change sinks. Keyed by subscription id for O(1) removal.
    pub(crate) topology_sinks: HashMap<u64, crate::topology::TopologySink>,
    pub(crate) next_topology_id: u64,
    /// A6 reentrancy guard (Slice F, 2026-05-07): the stack of NodeIds whose
    /// fn is currently being invoked on the wave-owner thread. Pushed at the
    /// top of `fire_fn` (just before the lock-released `invoke_fn` call) and
    /// popped on return / unwind via the [`crate::batch::FiringGuard`] RAII
    /// helper. [`Core::set_deps`] consults this set and rejects with
    /// [`SetDepsError::ReentrantOnFiringNode`] if `n` is currently firing —
    /// preventing the D1 `tracked` index corruption (see
    /// `porting-deferred.md` "Set_deps from inside firing node's fn corrupts
    /// Dynamic `tracked` indices").
    ///
    /// Stack rather than set so nested fn re-entrance (Producer subscribing
    /// to a fn that itself fires another fn) tracks every concurrently-firing
    /// node on the wave-owner. `Vec` rather than `HashSet` because the
    /// expected depth is small (typically 1, occasionally 2–3 with
    /// higher-order operators) and linear scan is faster than hash for that
    /// size.
    pub(crate) currently_firing: Vec<NodeId>,
    /// R1.3.8.c pause-overflow ERROR synthesis queue (Slice F, A3 —
    /// 2026-05-07). Recorded by [`Core::queue_notify`] when the pause
    /// buffer first overflows in a cycle; drained at wave-end after the
    /// lock-released call to
    /// [`crate::boundary::BindingBoundary::synthesize_pause_overflow_error`].
    ///
    /// One entry per (node × pause-cycle); subsequent overflows in the
    /// same cycle don't re-queue (gated by `PauseState::overflow_reported`).
    pub(crate) pending_pause_overflow: Vec<PendingPauseOverflow>,
    /// Slice G (R1.3.2.d / R1.3.3.a — 2026-05-07): nodes that have emitted
    /// at least one tier-3 message (Data or Resolved) in the CURRENT wave.
    /// Wave-scoped (cleared in `clear_wave_state`). Used by
    /// [`crate::batch::Core::commit_emission`] to detect "this is a
    /// subsequent emit at this node in the same wave" — when set,
    /// equals substitution is skipped (would produce a R1.3.3.a-violating
    /// mixed wave) and any prior Resolved entries in pending_notify or
    /// the pause buffer are rewritten to Data using the wave-start cache
    /// snapshot.
    ///
    /// Distinct from `pending_pause_overflow` (per-pause-cycle, not
    /// per-wave) and `wave_cache_snapshots` (per-wave snapshot, but only
    /// populated on Data path pre-Slice-G). Populated by both Data and
    /// Resolved branches of `commit_emission`; NOT populated by
    /// `commit_emission_verbatim` (Batch path passes through verbatim
    /// per R1.3.3.c).
    pub(crate) tier3_emitted_this_wave: ahash::AHashSet<NodeId>,
    /// Slice E2 (R1.3.9.b strict reading per D057): per-wave-per-node
    /// dedup for `OnInvalidate` cleanup hook firing. A node already in
    /// this set this wave has already had its `OnInvalidate` queued into
    /// `deferred_cleanup_hooks` and MUST NOT queue again, even if
    /// `invalidate_inner` re-encounters it (rare: only matters when the
    /// node re-populates mid-wave via fn-fire and then gets re-invalidated
    /// in the same wave through a separate path).
    ///
    /// Cleared in [`CoreState::clear_wave_state`] alongside the other
    /// wave-scoped queues.
    pub(crate) invalidate_hooks_fired_this_wave: ahash::AHashSet<NodeId>,
    /// Slice E2 (per D060/D061): lock-released drain queue for
    /// `OnInvalidate` cleanup hooks. Populated under the state lock by
    /// `Core::invalidate_inner` when a node's cache transitions
    /// `!= NO_HANDLE → NO_HANDLE`; drained after the lock drops at wave
    /// boundary by [`crate::batch::Core::fire_deferred_cleanup_hooks`]
    /// (each call wrapped in `catch_unwind` so a single binding panic
    /// doesn't short-circuit the drain — last panic re-raises after the
    /// loop completes per D060).
    ///
    /// **Panic-discard semantics (D061):** cleared in
    /// [`CoreState::clear_wave_state`] without firing — a panic-discarded
    /// wave drops the queued cleanup hooks silently, mirroring the
    /// `pending_pause_overflow` precedent (Slice F /qa A3). Bindings using
    /// `OnInvalidate` for external-resource cleanup MUST idempotent-cleanup
    /// at process exit or next successful invalidate cycle.
    pub(crate) deferred_cleanup_hooks: Vec<(NodeId, crate::boundary::CleanupTrigger)>,
    /// Slice E2 /qa Q2(b) (D069): lock-released drain queue for
    /// `BindingBoundary::wipe_ctx` calls fired eagerly from
    /// `Core::terminate_node` when a resubscribable node terminates with
    /// no live subscribers. Pairs with the `Subscription::Drop` direct-
    /// fire site (mutually exclusive: subs-empty-at-terminate routes
    /// here; subs-non-empty-at-terminate fires from Subscription::Drop's
    /// last-sub-drop path). Drained alongside `deferred_cleanup_hooks`
    /// at wave boundary; same `catch_unwind` discipline so a single
    /// binding panic doesn't short-circuit the drain. Same panic-discard
    /// semantics as `deferred_cleanup_hooks` (silent drop on
    /// panic-discarded waves).
    pub(crate) pending_wipes: Vec<NodeId>,
}

/// The handle-protocol Core dispatcher.
///
/// Holds an [`Arc`] to the [`BindingBoundary`] and all dispatch state. Cheap
/// to clone (the inner `Arc<Mutex<CoreState>>` is shared); pass `Core` by
/// value to threads.
///
/// # Wave-owner re-entrant mutex (Slice A close /qa, M1)
///
/// The state lock (`state: Mutex<CoreState>`) is **dropped** around binding
/// callbacks (`invoke_fn`, `custom_equals`) so user fns may re-enter Core.
/// To preserve serializability of WAVE EXECUTION across threads — without
/// re-introducing the lock-held-during-fn-fire deadlock the Slice A close
/// refactor lifted — the wave engine acquires `wave_owner` (a
/// [`parking_lot::ReentrantMutex`]) for the lifetime of each wave.
///
/// Properties:
///
/// - **Same-thread re-entrance is free.** A user fn that calls back into
///   `Core::emit` / `Core::pause` / etc. mid-fire re-acquires `wave_owner`
///   on the same thread and runs as a nested wave (the inner `run_wave`
///   sees `in_tick=true` and skips drain — outer drain picks up).
/// - **Cross-thread emits BLOCK** at `wave_owner.lock_arc()` until the
///   in-flight wave completes (drain + flush + sink fire all done). This
///   serializes wave OWNERSHIP across threads, while still allowing the
///   state lock to drop inside the wave for binding callbacks.
///
/// Without this, Slice A close's lock-released drain let cross-thread
/// emits absorb into the in-flight wave's `pending_notify` and return
/// before subscribers fire — breaking the user-facing happens-after
/// contract that `emit` returning means subscribers have observed.
#[derive(Clone)]
pub struct Core {
    pub(crate) state: Arc<Mutex<CoreState>>,
    pub(crate) binding: Arc<dyn BindingBoundary>,
    pub(crate) wave_owner: Arc<ReentrantMutex<()>>,
    /// Slice X5 (D3 substrate, 2026-05-08): per-subgraph union-find
    /// registry. Tracks each registered node's connected-component
    /// membership (a "subgraph") so cross-thread emits to disjoint
    /// components can run truly parallel via per-component
    /// `wave_owner` (Y1 commit-2 wires the wave engine through the
    /// registry; X5 commit-1 just maintains the union-find state).
    ///
    /// Direct port of [`graphrefly-py`'s
    /// `subgraph_locks.py`](https://github.com/graphrefly/graphrefly-py/blob/main/src/graphrefly/core/subgraph_locks.py)
    /// design (locked in [`SESSION-rust-port-d3-per-subgraph-parallelism.md`](https://github.com/graphrefly/graphrefly-ts/blob/main/archive/docs/SESSION-rust-port-d3-per-subgraph-parallelism.md)).
    pub(crate) registry: Arc<parking_lot::Mutex<crate::subgraph::SubgraphRegistry>>,
}

/// Weak handle to a [`Core`] — does not contribute to strong refcount.
///
/// Constructed via [`Core::weak_handle`]; upgraded back to a strong
/// [`Core`] via [`WeakCore::upgrade`]. Used by long-lived binding-stored
/// closures (notably `ProducerBuildFn`s registered via
/// [`graphrefly_operators::ProducerBinding::register_producer_build`])
/// to break the BenchBinding → registry → closure → strong-Core cycle
/// that would otherwise leak the entire graph state when a `BenchCore`
/// drops with active producer registrations.
///
/// Upgrade on each invocation; if the host `Core` was already dropped,
/// `upgrade()` returns `None` and the closure should no-op (the host
/// is being torn down, no work to do).
#[derive(Clone)]
pub struct WeakCore {
    state: Weak<Mutex<CoreState>>,
    binding: Weak<dyn BindingBoundary>,
    wave_owner: Weak<ReentrantMutex<()>>,
    registry: Weak<parking_lot::Mutex<crate::subgraph::SubgraphRegistry>>,
}

impl WeakCore {
    /// Try to upgrade back to a strong [`Core`]. Returns `None` if the
    /// host `Core`'s strong count has reached zero (i.e. the host
    /// `BenchCore` / equivalent owner was dropped).
    #[must_use]
    pub fn upgrade(&self) -> Option<Core> {
        Some(Core {
            state: self.state.upgrade()?,
            binding: self.binding.upgrade()?,
            wave_owner: self.wave_owner.upgrade()?,
            registry: self.registry.upgrade()?,
        })
    }
}

/// RAII guard that owns an [`OperatorScratch`] until either (a) the
/// caller `take()`s it for installation, or (b) the guard drops on an
/// early return / unwind, in which case the scratch's handle retains
/// are released via [`OperatorScratch::release_handles`].
///
/// Slice H /qa F1 + F2 (2026-05-07): closes two related correctness
/// gaps in `Core::register`:
///
/// 1. **TOCTOU window** — the original three-phase split called
///    `lock_state()` twice (once for validation, once for insertion),
///    so a concurrent `Core::complete(dep)` on a non-resubscribable
///    dep could slip in between the two acquisitions and re-create
///    the wedge `RegisterError::TerminalDep` was designed to prevent.
///    The guard plus a single locked region for both phases closes
///    this gap (release runs lock-released because guard variables
///    drop in reverse declaration order — guard declared BEFORE
///    `lock_state()`, so the lock guard drops first).
///
/// 2. **Panic-unsafe scratch leak** — without an RAII drop, a panic
///    between `make_op_scratch` (Phase 2) and the explicit
///    `if let Err(e)` cleanup branch (e.g., `lock_state()` reentrance
///    assert, OOM-as-panic on Vec growth in dep iteration) would
///    drop the `Box<dyn OperatorScratch>` without releasing the
///    seed/default retain. The guard's `Drop` impl releases on any
///    unwind path.
///
/// Lock-discipline: the guard holds `&dyn BindingBoundary` (through
/// the `Arc<dyn BindingBoundary>` it borrows from). On `Drop`, it
/// invokes `release_handles` lock-released — fires AFTER any
/// `MutexGuard<CoreState>` declared later in the same scope drops
/// (LIFO destruction order). Mirrors `Core::resume` Phase 2 release
/// pattern.
struct ScratchReleaseGuard<'a> {
    scratch: Option<Box<dyn crate::op_state::OperatorScratch>>,
    binding: &'a dyn BindingBoundary,
}

impl<'a> ScratchReleaseGuard<'a> {
    fn new(
        scratch: Option<Box<dyn crate::op_state::OperatorScratch>>,
        binding: &'a dyn BindingBoundary,
    ) -> Self {
        Self { scratch, binding }
    }

    /// Take ownership of the scratch — disarms the release-on-drop
    /// behavior. Used on the success path to install the scratch on
    /// `NodeRecord.op_scratch`.
    fn take(mut self) -> Option<Box<dyn crate::op_state::OperatorScratch>> {
        self.scratch.take()
    }
}

impl Drop for ScratchReleaseGuard<'_> {
    fn drop(&mut self) {
        if let Some(mut scratch) = self.scratch.take() {
            scratch.release_handles(self.binding);
        }
    }
}

impl Core {
    /// Construct a fresh Core wired to the given binding. Pause buffer cap
    /// defaults to unbounded; set via [`Self::set_pause_buffer_cap`].
    #[must_use]
    pub fn new(binding: Arc<dyn BindingBoundary>) -> Self {
        Self {
            state: Arc::new(Mutex::new(CoreState {
                next_node_id: 1,
                next_subscription_id: 1,
                // A4 (Slice F, 2026-05-07): start `next_lock_id` in the high
                // half of the u32 range so `alloc_lock_id` can't collide with
                // user-supplied `LockId::new(N)` constructors (which the
                // napi-rs binding marshals from `u32` and tests typically use
                // in the low range, 1..1024). Phase E /qa F1 (2026-05-08):
                // lowered from `1u64 << 32` to `1u64 << 31` so the value
                // round-trips through `u32::try_from(...)` at the napi
                // boundary — the previous seed errored every napi
                // `alloc_lock_id` call. Anti-collision intent (high range vs
                // low user range) preserved at half the prior ceiling
                // (2^31 ≈ 2 billion allocations per Core, ample for parity
                // tests). Lift the floor when the deferred BigInt-narrowing
                // migration extends `LockId` to `u64` at the FFI layer
                // (porting-deferred "BigInt migration for u32-narrowed napi
                // types" entry).
                next_lock_id: 1u64 << 31,
                nodes: HashMap::new(),
                children: HashMap::new(),
                pending_fires: HashSet::new(),
                pending_notify: IndexMap::new(),
                in_tick: false,
                pause_buffer_cap: None,
                max_batch_drain_iterations: 10_000,
                deferred_flush_jobs: Vec::new(),
                deferred_handle_releases: Vec::new(),
                binding: binding.clone(),
                wave_cache_snapshots: HashMap::new(),
                pending_auto_resolve: ahash::AHashSet::new(),
                topology_sinks: HashMap::new(),
                next_topology_id: 1,
                currently_firing: Vec::new(),
                pending_pause_overflow: Vec::new(),
                tier3_emitted_this_wave: ahash::AHashSet::new(),
                invalidate_hooks_fired_this_wave: ahash::AHashSet::new(),
                deferred_cleanup_hooks: Vec::new(),
                pending_wipes: Vec::new(),
            })),
            binding,
            wave_owner: Arc::new(ReentrantMutex::new(())),
            registry: Arc::new(parking_lot::Mutex::new(
                crate::subgraph::SubgraphRegistry::new(),
            )),
        }
    }

    /// Acquire the state lock.
    ///
    /// Post-Slice-E: `Core::subscribe` fires the per-tier handshake
    /// LOCK-RELEASED with `wave_owner` held; sink callbacks may freely
    /// re-enter Core (`emit` / `complete` / `error` / nested `subscribe`).
    /// Same-thread re-entry passes through `wave_owner`'s `ReentrantMutex`
    /// transparently; cross-thread emits block on `wave_owner` until the
    /// outer subscribe completes, preserving R1.3.5.a happens-after
    /// ordering. The previous `IN_HANDSHAKE_FIRE` panic-diagnostic is no
    /// longer needed.
    pub(crate) fn lock_state(&self) -> MutexGuard<'_, CoreState> {
        self.state.lock()
    }

    /// Whether `self` and `other` point to the same dispatcher state.
    /// True when one was produced by `Clone`-ing the other (or they
    /// were both cloned from a common ancestor); false for two
    /// independently `Core::new`-constructed instances even with the
    /// same binding.
    ///
    /// Used by `graphrefly-graph`'s `mount` to enforce the "shared-Core
    /// only" v1 invariant — cross-Core mount is post-M6.
    #[must_use]
    pub fn same_dispatcher(&self, other: &Core) -> bool {
        Arc::ptr_eq(&self.state, &other.state)
    }

    /// Downgrade to a [`WeakCore`] handle that doesn't contribute to
    /// strong refcount of the underlying state / binding / wave_owner.
    ///
    /// Used by binding-stored long-lived closures (e.g.
    /// `register_producer_build`-stored `ProducerBuildFn`s) to avoid the
    /// Arc cycle:
    ///
    /// ```text
    /// BenchBinding → registry → producer_builds[fn_id]
    ///   → closure → strong Arc<dyn _Binding> → BenchBinding
    /// ```
    ///
    /// Closures hold `WeakCore` and `Weak<dyn _Binding>` instead, then
    /// upgrade-on-fire (returning early if either weak is dangling —
    /// indicating the host BenchCore was already dropped). Upgraded
    /// strong refs live only for the build closure's invocation; sinks
    /// the build closure spawns close over those upgraded strongs and
    /// stay alive only while the producer is active (cleared via
    /// `producer_deactivate` on last-subscriber unsubscribe).
    #[must_use]
    pub fn weak_handle(&self) -> WeakCore {
        WeakCore {
            state: Arc::downgrade(&self.state),
            binding: Arc::downgrade(&self.binding),
            wave_owner: Arc::downgrade(&self.wave_owner),
            registry: Arc::downgrade(&self.registry),
        }
    }

    /// Number of distinct connected-component partitions tracked by
    /// the per-subgraph union-find registry (Slice X5 substrate).
    /// Two threads emitting into nodes with distinct partitions will
    /// run truly parallel once Y1 wires the wave engine through the
    /// registry; X5 reports the partition count for inspection
    /// (acceptance bar + debugging) but the wave engine still uses
    /// the legacy Core-level `wave_owner`.
    #[must_use]
    pub fn partition_count(&self) -> usize {
        self.registry.lock().component_count()
    }

    /// Resolve `node`'s partition identity per the per-subgraph
    /// union-find registry (Slice X5 substrate). Two nodes with the
    /// same `SubgraphId` are connected via dep edges (transitively)
    /// and share a partition lock under Y1+; nodes in different
    /// partitions can run truly parallel.
    ///
    /// Returns `None` for unregistered nodes.
    #[must_use]
    pub fn partition_of(&self, node: NodeId) -> Option<crate::subgraph::SubgraphId> {
        self.registry.lock().partition_of(node)
    }

    /// Test-only inspection: number of `PendingBatch`es queued for
    /// `node` in the current wave. Used by Slice X4 D2 regression
    /// tests to pin the "common case = single batch, no SmallVec
    /// spill" perf invariant.
    ///
    /// Returns `None` if no `pending_notify` entry exists for `node`
    /// (no tier-1+ message has been queued for this node yet in this
    /// wave). `Some(0)` is unreachable by construction (a vacant
    /// entry implies no batches; an occupied entry has at least one).
    #[cfg(any(test, debug_assertions))]
    #[must_use]
    pub fn pending_batch_count(&self, node: NodeId) -> Option<usize> {
        self.lock_state()
            .pending_notify
            .get(&node)
            .map(|entry| entry.batches.len())
    }

    /// Configure the Core-global cap on pause replay buffer length. When set,
    /// any per-node pause buffer that would exceed `cap` drops the oldest
    /// message(s) from the front; the dropped count is reported back via the
    /// resume callback (see [`ResumeReport`]). `None` (default) means
    /// unbounded; messages buffer indefinitely until the lockset clears.
    pub fn set_pause_buffer_cap(&self, cap: Option<usize>) {
        self.lock_state().pause_buffer_cap = cap;
    }

    /// Configure the replay buffer cap on `node_id` (R2.6.5 / Lock 6.G —
    /// Slice E1, 2026-05-07). `None` disables the buffer. `Some(N)` keeps
    /// the last `N` DATA emissions in a circular buffer; late subscribers
    /// receive them as part of the per-tier handshake (between START and
    /// any terminal). Switching from a larger cap to a smaller cap evicts
    /// the front of the buffer to fit; switching to `None` drains the
    /// buffer entirely. Each evicted/drained handle's retain is released
    /// back to the binding.
    ///
    /// # Panics
    ///
    /// Panics if `node_id` is not registered.
    pub fn set_replay_buffer_cap(&self, node_id: NodeId, cap: Option<usize>) {
        // QA A7 (2026-05-07): normalize `Some(0)` to `None`. Two ways to
        // express "disabled" is confusing: `push_replay_buffer` already
        // treats `Some(0)` as no-op, so persisting it adds nothing.
        let cap = match cap {
            Some(0) => None,
            other => other,
        };
        let to_release: Vec<HandleId> = {
            let mut s = self.lock_state();
            let rec = s.require_node_mut(node_id);
            rec.replay_buffer_cap = cap;
            match cap {
                None => rec.replay_buffer.drain(..).collect(),
                Some(c) => {
                    let mut drained = Vec::new();
                    while rec.replay_buffer.len() > c {
                        if let Some(h) = rec.replay_buffer.pop_front() {
                            drained.push(h);
                        }
                    }
                    drained
                }
            }
        };
        for h in to_release {
            self.binding.release_handle(h);
        }
    }

    /// Reconfigure the pause mode for `node_id` (canonical §2.6 — Slice F
    /// audit close, 2026-05-07). Default for new nodes is
    /// [`PausableMode::Default`]; switch to [`PausableMode::ResumeAll`]
    /// for nodes whose pause-window emit history must be observable
    /// verbatim, or [`PausableMode::Off`] for nodes intrinsically
    /// pause-immune.
    ///
    /// # Errors
    ///
    /// - [`SetPausableModeError::UnknownNode`] — `node_id` is not
    ///   registered.
    /// - [`SetPausableModeError::WhilePaused`] — the node currently
    ///   holds at least one pause lock. Changing mode mid-pause would
    ///   lose buffered content or strand a `pending_wave` flag — resume
    ///   all locks first.
    pub fn set_pausable_mode(
        &self,
        node_id: NodeId,
        mode: PausableMode,
    ) -> Result<(), SetPausableModeError> {
        let mut s = self.lock_state();
        let rec = s
            .nodes
            .get_mut(&node_id)
            .ok_or(SetPausableModeError::UnknownNode(node_id))?;
        if rec.pause_state.is_paused() {
            return Err(SetPausableModeError::WhilePaused);
        }
        rec.pausable = mode;
        Ok(())
    }

    /// Configure the wave-drain iteration cap (R4.3 / Lock 2.F′). The wave
    /// engine aborts a drain after `cap` iterations with a diagnostic panic.
    /// Default is `10_000` — high enough to avoid false positives on legitimate
    /// fan-in cascades, low enough to surface runtime cycles within seconds.
    ///
    /// Lower this only when running adversarial / property-based tests that
    /// want fast cycle detection. Raise it only with concrete evidence that a
    /// legitimate workload needs more iterations than the default — and even
    /// then, prefer to tune the workload (per-subgraph batching, etc.) over
    /// raising the cap.
    ///
    /// # Panics
    ///
    /// Panics if `cap == 0` — a zero cap would abort every wave on the very
    /// first iteration, deadlocking any subsequent dispatcher work.
    pub fn set_max_batch_drain_iterations(&self, cap: u32) {
        assert!(cap > 0, "max_batch_drain_iterations must be > 0");
        self.lock_state().max_batch_drain_iterations = cap;
    }

    /// Send a message UPSTREAM from `node_id` to each of its declared deps
    /// (canonical R1.4.1 — Slice F audit, F2 / 2026-05-07).
    ///
    /// The dispatcher rejects tier-3 (DATA / RESOLVED) and tier-5
    /// (COMPLETE / ERROR) per R1.4.1: value and terminal-lifecycle planes
    /// are downstream-only. All other tiers (0 START, 1 DIRTY, 2 PAUSE /
    /// RESUME, 4 INVALIDATE, 6 TEARDOWN) pass.
    ///
    /// # Routing per tier
    ///
    /// - **Tier 0 ([`Message::Start`]):** no-op. START is a per-subscription
    ///   handshake, not a routable wire signal — sending it upstream has no
    ///   well-defined target.
    /// - **Tier 1 ([`Message::Dirty`]):** no-op. The dep's "something
    ///   changed" notification is its own [`Self::emit`] / commit
    ///   responsibility; ignoring upstream DIRTY hints is safe.
    /// - **Tier 2 ([`Message::Pause`] / [`Message::Resume`]):** translates
    ///   to [`Self::pause`] / [`Self::resume`] on each dep. Lock id is
    ///   forwarded verbatim. Errors from individual deps are accumulated
    ///   in the `dep_errors` field of the returned report.
    /// - **Tier 4 ([`Message::Invalidate`]):** translates to
    ///   [`Self::invalidate`] on each dep. Note: canonical R1.4.2
    ///   distinguishes "downstream INVALIDATE" (cache clear + cascade) from
    ///   "upstream INVALIDATE" (plain forward, no self-process). The Rust
    ///   port v1 SIMPLIFICATION delegates to the same `Core::invalidate`
    ///   path — upstream INVALIDATE here DOES clear dep caches and cascade.
    ///   If a "plain forward" mode surfaces as a real consumer need, add
    ///   `up_with_options`.
    /// - **Tier 6 ([`Message::Teardown`]):** translates to
    ///   [`Self::teardown`] on each dep. Cascades per the standard
    ///   teardown path.
    ///
    /// # Errors
    ///
    /// - [`UpError::UnknownNode`] — `node_id` is not registered.
    /// - [`UpError::TierForbidden`] — tier 3 or tier 5.
    pub fn up(&self, node_id: NodeId, message: Message) -> Result<(), UpError> {
        // QA A10 (2026-05-07): check unknown node BEFORE tier rejection
        // for consistent error UX — `up(unknown, Data)` and
        // `up(unknown, Pause)` both report `UnknownNode` rather than
        // splitting on the tier.
        let dep_ids: Vec<NodeId> = {
            let s = self.lock_state();
            let rec = s.nodes.get(&node_id).ok_or(UpError::UnknownNode(node_id))?;
            rec.dep_ids_vec()
        };
        let tier = message.tier();
        if tier == 3 || tier == 5 {
            return Err(UpError::TierForbidden { tier });
        }
        for dep_id in dep_ids {
            match message {
                Message::Pause(lock) => {
                    let _ = self.pause(dep_id, lock);
                }
                Message::Resume(lock) => {
                    let _ = self.resume(dep_id, lock);
                }
                Message::Invalidate => {
                    self.invalidate(dep_id);
                }
                Message::Teardown => {
                    self.teardown(dep_id);
                }
                // Tier 0 START + tier 1 DIRTY: no-op upstream per the
                // routing table above.
                _ => {}
            }
        }
        Ok(())
    }

    /// Allocate a unique [`LockId`] for use with [`Self::pause`] /
    /// [`Self::resume`]. Convenience for callers that don't already have an
    /// id-allocation scheme; user-supplied ids work too.
    #[must_use]
    pub fn alloc_lock_id(&self) -> LockId {
        let mut s = self.lock_state();
        let id = LockId::new(s.next_lock_id);
        s.next_lock_id += 1;
        id
    }

    // -------------------------------------------------------------------
    // Registration — unified `register()` (D030, Slice D)
    //
    // All node kinds (State / Producer / Derived / Dynamic / Operator)
    // funnel through `Core::register(NodeRegistration) -> NodeId`. Sugar
    // wrappers (`register_state` / `register_producer` / `register_derived`
    // / `register_dynamic` / `register_operator`) build a `NodeRegistration`
    // and delegate. There is no parallel registration path internally.
    // -------------------------------------------------------------------

    /// Unified node registration (D030).
    ///
    /// `reg` describes the node's identity (deps + closure-form fn id OR
    /// typed-op + per-kind opts). The kind is **derived from the field
    /// shape**, not stored — see [`NodeKind`].
    ///
    /// Sugar wrappers below ([`Self::register_state`],
    /// [`Self::register_producer`], [`Self::register_derived`],
    /// [`Self::register_dynamic`], [`Self::register_operator`]) build the
    /// registration for the common kinds and delegate here. Direct callers
    /// that need uncommon combinations (e.g., a partial-true derived) can
    /// invoke this method directly.
    ///
    /// # Errors
    ///
    /// Errors are returned in evaluation order — earlier phases short-circuit
    /// later ones, so a single registration produces at most one variant.
    ///
    /// **Phase 1 — lock-released, side-effect-free validation:**
    /// - [`RegisterError::OperatorWithoutDeps`] — `reg` carries an op but
    ///   `deps` is empty. Operator nodes need at least one dep — for
    ///   subscription-managed combinators with no declared deps, use
    ///   [`Self::register_producer`] instead.
    /// - [`RegisterError::InitialOnlyForStateNodes`] — `reg.opts.initial`
    ///   is non-sentinel for a non-state shape (deps non-empty, or
    ///   fn_or_op present). State nodes are the only kind with an initial
    ///   cache.
    ///
    /// **Phase 2 — operator scratch construction (lock-released):**
    /// - [`RegisterError::OperatorSeedSentinel`] — `reg` carries `Op(Scan)`
    ///   / `Op(Reduce)` with a `NO_HANDLE` seed. R2.5.3 — stateful folders
    ///   must have a real seed.
    ///
    /// **Phase 3 — state-lock validation (folded with insertion under a
    /// single lock acquisition per /qa F1 to prevent TOCTOU between
    /// validation and `nodes.insert`):**
    /// - [`RegisterError::UnknownDep`] — any element of `reg.deps` is not
    ///   a registered node id.
    /// - [`RegisterError::TerminalDep`] — a dep is terminal (COMPLETE /
    ///   ERROR) AND not resubscribable — would create a permanent wedge.
    ///
    /// All errors are construction-time invariants — the dispatcher
    /// rejects the registration before any reactive state is created.
    /// On `Err`, no node has been added and any handle retains taken on
    /// the way in (operator scratch seed retains via
    /// [`BindingBoundary::retain_handle`]) have been released
    /// lock-released — see [`ScratchReleaseGuard`] for the RAII
    /// discipline that covers both early-return AND unwind paths.
    /// `Last { default }` retains its `default` handle on the same
    /// release path.
    pub fn register(&self, reg: NodeRegistration) -> Result<NodeId, RegisterError> {
        let NodeRegistration {
            deps,
            fn_or_op,
            opts,
        } = reg;
        let NodeOpts {
            initial,
            equals,
            partial,
            is_dynamic,
            pausable,
            replay_buffer,
        } = opts;

        // Derive the field shape from fn_or_op + deps.
        let (fn_id, op) = match fn_or_op {
            Some(NodeFnOrOp::Fn(f)) => (Some(f), None),
            Some(NodeFnOrOp::Op(o)) => (None, Some(o)),
            None => (None, None),
        };

        // Phase 1 — lock-released, side-effect-free validation. Errors
        // here return BEFORE any handle retain is taken.
        //
        //   - State (no deps + no fn + no op) is the only kind with `initial`.
        //   - Dynamic flag only meaningful when fn + non-empty deps.
        //   - Operator (op present) must have deps (P9: operator without deps
        //     would skip activation — use a producer instead).
        let is_state_shape = deps.is_empty() && fn_id.is_none() && op.is_none();
        if op.is_some() && deps.is_empty() {
            return Err(RegisterError::OperatorWithoutDeps);
        }
        if initial != NO_HANDLE && !is_state_shape {
            return Err(RegisterError::InitialOnlyForStateNodes);
        }

        // Phase 2 — build per-operator scratch struct (may take handle
        // retains via `binding.retain_handle` for Scan/Reduce/Last seed).
        // Lock-released per Slice E (D045) handshake discipline. Returns
        // `OperatorSeedSentinel` BEFORE retain so an Err leaves no
        // dangling handles.
        let scratch = match op {
            Some(operator_op) => self.make_op_scratch(operator_op)?,
            None => None,
        };

        // Wrap scratch in an RAII guard immediately after Phase 2. From
        // here on, ANY early return / unwind path correctly releases the
        // scratch's handle retains via `OperatorScratch::release_handles`
        // (Slice H /qa F2 — defense against panics between Phase 2 and
        // Phase 3 cleanup branch). Lock-released because the guard is
        // declared BEFORE `lock_state()` below — variable destruction
        // order is reverse declaration order, so the `MutexGuard` drops
        // first on any return path.
        let scratch_guard = ScratchReleaseGuard::new(scratch, &*self.binding);

        // Phase 3 — state-lock-required validation, FOLDED with insertion
        // under a single `lock_state()` acquisition per /qa F1. The
        // pre-/qa version split this into two acquisitions (one for
        // validation, one for `alloc_node_id` + `nodes.insert`), opening
        // a TOCTOU window where a concurrent `Core::complete(dep)` on a
        // non-resubscribable dep could slip in and recreate the wedge
        // `TerminalDep` was designed to prevent. Single locked region
        // closes the gap.
        let mut s = self.lock_state();

        for &dep in &deps {
            if !s.nodes.contains_key(&dep) {
                return Err(RegisterError::UnknownDep(dep));
            }
        }
        // Slice F audit (2026-05-07): mirror `set_deps`'s `TerminalDep`
        // rejection at registration time. Adding a non-resubscribable
        // terminal node as a dep at registration creates a permanent wedge.
        for &dep in &deps {
            let dep_rec = s.require_node(dep);
            if dep_rec.terminal.is_some() && !dep_rec.resubscribable {
                return Err(RegisterError::TerminalDep(dep));
            }
        }

        // Validation passed — install. Take scratch out of the guard
        // (disarms the release-on-drop) and continue using `s`.
        let installed_scratch = scratch_guard.take();

        let id = s.alloc_node_id();

        // `tracked`: Static derived + Operator track all deps; Dynamic
        // starts empty and fills via fn return; State / Producer have no
        // deps so tracked is empty.
        let tracked: HashSet<usize> = if op.is_some() {
            (0..deps.len()).collect()
        } else if is_dynamic {
            HashSet::new()
        } else if fn_id.is_some() && !deps.is_empty() {
            // Static derived
            (0..deps.len()).collect()
        } else {
            HashSet::new()
        };

        let dep_records: Vec<DepRecord> = deps.iter().map(|&d| DepRecord::new(d)).collect();

        let rec = NodeRecord {
            dep_records,
            fn_id,
            op,
            is_dynamic,
            equals,
            cache: initial,
            has_fired_once: initial != NO_HANDLE,
            subscribers: HashMap::new(),
            subscribers_revision: 0,
            tracked,
            dirty: false,
            involved_this_wave: false,
            pause_state: PauseState::Active,
            pausable,
            replay_buffer_cap: replay_buffer,
            replay_buffer: VecDeque::new(),
            terminal: None,
            has_received_teardown: false,
            resubscribable: false,
            meta_companions: Vec::new(),
            partial,
            op_scratch: installed_scratch,
        };
        s.nodes.insert(id, rec);
        s.children.insert(id, HashSet::new());
        for &dep in &deps {
            s.children.entry(dep).or_default().insert(id);
        }
        drop(s);
        // Slice X5 (D3 substrate, 2026-05-08): track partition membership.
        // Register the new node as its own component, then `union_nodes`
        // each dep — connectivity-based grouping per Q1=(c-uf split-eager).
        // Lock-released wrt state so the registry mutex isn't ordered
        // under the state mutex (avoids constraining future cross-mutex
        // refactors).
        {
            let mut reg = self.registry.lock();
            reg.ensure_registered(id);
            for &dep in &deps {
                reg.union_nodes(id, dep);
            }
        }
        self.fire_topology_event(&crate::topology::TopologyEvent::NodeRegistered(id));
        Ok(id)
    }

    /// Sugar over [`Self::register`] — register a state node. `initial`
    /// may be [`NO_HANDLE`] to start sentinel.
    ///
    /// `partial` is accepted for surface consistency (D019); for state
    /// nodes it has no effect (state nodes don't fire fn).
    ///
    /// # Errors
    ///
    /// State registration is structurally simple — no deps, no op — so
    /// the only reachable variant is none in practice. Returns
    /// [`Result`] for surface consistency with [`Self::register`].
    pub fn register_state(
        &self,
        initial: HandleId,
        partial: bool,
    ) -> Result<NodeId, RegisterError> {
        self.register(NodeRegistration {
            deps: Vec::new(),
            fn_or_op: None,
            opts: NodeOpts {
                initial,
                partial,
                ..NodeOpts::default()
            },
        })
    }

    /// Sugar over [`Self::register`] — register a producer node (D031,
    /// Slice D). No deps; fn fires once on first subscribe; cleanup runs
    /// via [`BindingBoundary::producer_deactivate`] when the last
    /// subscriber unsubscribes.
    ///
    /// The fn body uses the binding's `ProducerCtx`-equivalent helper
    /// (see `graphrefly-operators::producer`) to subscribe to other Core
    /// nodes — the zip / concat / race / takeUntil pattern.
    ///
    /// # Errors
    ///
    /// Producer registration has no user-supplied deps, so structurally
    /// none of [`RegisterError`]'s variants are reachable. Returns
    /// [`Result`] for surface consistency with [`Self::register`].
    pub fn register_producer(&self, fn_id: FnId) -> Result<NodeId, RegisterError> {
        self.register(NodeRegistration {
            deps: Vec::new(),
            fn_or_op: Some(NodeFnOrOp::Fn(fn_id)),
            opts: NodeOpts {
                // Producers have no deps — the first-run gate is degenerate.
                partial: true,
                ..NodeOpts::default()
            },
        })
    }

    /// Sugar over [`Self::register`] — register a derived (static) node.
    /// `partial` controls the R2.5.3 first-run gate (D011).
    ///
    /// # Errors
    ///
    /// - [`RegisterError::UnknownDep`] — any element of `deps` is not
    ///   registered.
    /// - [`RegisterError::TerminalDep`] — a dep is terminal and not
    ///   resubscribable.
    pub fn register_derived(
        &self,
        deps: &[NodeId],
        fn_id: FnId,
        equals: EqualsMode,
        partial: bool,
    ) -> Result<NodeId, RegisterError> {
        self.register(NodeRegistration {
            deps: deps.to_vec(),
            fn_or_op: Some(NodeFnOrOp::Fn(fn_id)),
            opts: NodeOpts {
                equals,
                partial,
                ..NodeOpts::default()
            },
        })
    }

    /// Sugar over [`Self::register`] — register a dynamic node (fn
    /// declares its actually-tracked dep indices per fire). `partial`
    /// controls the R2.5.3 first-run gate (D011).
    ///
    /// # Errors
    ///
    /// - [`RegisterError::UnknownDep`] — any element of `deps` is not
    ///   registered.
    /// - [`RegisterError::TerminalDep`] — a dep is terminal and not
    ///   resubscribable.
    pub fn register_dynamic(
        &self,
        deps: &[NodeId],
        fn_id: FnId,
        equals: EqualsMode,
        partial: bool,
    ) -> Result<NodeId, RegisterError> {
        self.register(NodeRegistration {
            deps: deps.to_vec(),
            fn_or_op: Some(NodeFnOrOp::Fn(fn_id)),
            opts: NodeOpts {
                equals,
                partial,
                is_dynamic: true,
                ..NodeOpts::default()
            },
        })
    }

    /// Build a fresh [`OperatorScratch`](crate::op_state::OperatorScratch)
    /// box for an operator variant, taking any required handle retains.
    /// Shared between `register_operator` (initial install) and
    /// `reset_for_fresh_lifecycle` (resubscribable cycle re-install).
    ///
    /// # Errors
    ///
    /// Returns [`RegisterError::OperatorSeedSentinel`] if `op` is `Scan`
    /// / `Reduce` with a [`NO_HANDLE`] seed (R2.5.3 — stateful folders
    /// must have a real seed). Refcount discipline: the seed-sentinel
    /// check happens BEFORE [`BindingBoundary::retain_handle`], so an
    /// `Err` leaves no handles dangling.
    fn make_op_scratch(
        &self,
        op: OperatorOp,
    ) -> Result<Option<Box<dyn crate::op_state::OperatorScratch>>, RegisterError> {
        use crate::op_state::{
            DistinctState, LastState, PairwiseState, ReduceState, ScanState, SkipState, TakeState,
            TakeWhileState,
        };
        // Slice H (2026-05-07): Scan/Reduce seed-sentinel checks happen
        // BEFORE retain_handle so an Err return leaves no handles dangling.
        //
        // Slice H /qa F13 (2026-05-07): for retaining variants, allocate
        // the `Box<State>` BEFORE calling `binding.retain_handle`. If
        // `Box::new` panics (e.g., OOM-as-panic), no retain has happened
        // yet — no leak. If `retain_handle` panics after Box succeeds,
        // the `Box<State>` is dropped on unwind; State has no handle yet
        // (we haven't touched the registry refcount), so still no leak.
        // Caller wraps the returned scratch in `ScratchReleaseGuard` to
        // cover panics AFTER make_op_scratch returns.
        match op {
            OperatorOp::Scan { seed, .. } => {
                if seed == NO_HANDLE {
                    return Err(RegisterError::OperatorSeedSentinel);
                }
                let state = Box::new(ScanState { acc: seed });
                self.binding.retain_handle(seed);
                Ok(Some(state))
            }
            OperatorOp::Reduce { seed, .. } => {
                if seed == NO_HANDLE {
                    return Err(RegisterError::OperatorSeedSentinel);
                }
                let state = Box::new(ReduceState { acc: seed });
                self.binding.retain_handle(seed);
                Ok(Some(state))
            }
            OperatorOp::DistinctUntilChanged { .. } => Ok(Some(Box::new(DistinctState::default()))),
            OperatorOp::Pairwise { .. } => Ok(Some(Box::new(PairwiseState::default()))),
            OperatorOp::Take { .. } => Ok(Some(Box::new(TakeState::default()))),
            OperatorOp::Skip { .. } => Ok(Some(Box::new(SkipState::default()))),
            OperatorOp::TakeWhile { .. } => Ok(Some(Box::new(TakeWhileState))),
            OperatorOp::Last { default } => {
                let state = Box::new(LastState {
                    latest: NO_HANDLE,
                    default,
                });
                if default != NO_HANDLE {
                    self.binding.retain_handle(default);
                }
                Ok(Some(state))
            }
            OperatorOp::Map { .. }
            | OperatorOp::Filter { .. }
            | OperatorOp::Combine { .. }
            | OperatorOp::WithLatestFrom { .. }
            | OperatorOp::Merge => Ok(None),
        }
    }

    /// Sugar over [`Self::register`] — register a built-in operator node
    /// (Slice C-1, D009; D026 generic scratch). The operator dispatch path
    /// lives in `fire_operator`; `op` selects which per-operator FFI
    /// method on [`BindingBoundary`] gets called per fire.
    ///
    /// For stateful operators ([`OperatorOp::Scan`] / [`Reduce`] /
    /// [`Last`] with a default), the seed/default handle is captured
    /// into the appropriate
    /// [`OperatorScratch`](crate::op_state::OperatorScratch) struct
    /// stored at [`NodeRecord::op_scratch`], and Core takes one retain
    /// share via [`BindingBoundary::retain_handle`].
    ///
    /// # Errors
    ///
    /// - [`RegisterError::OperatorWithoutDeps`] — `deps` is empty (use
    ///   [`Self::register_producer`] instead).
    /// - [`RegisterError::OperatorSeedSentinel`] — `op` is
    ///   [`OperatorOp::Scan`] / [`OperatorOp::Reduce`] with a
    ///   [`NO_HANDLE`] seed.
    /// - [`RegisterError::UnknownDep`] — any element of `deps` is not
    ///   registered.
    /// - [`RegisterError::TerminalDep`] — a dep is terminal and not
    ///   resubscribable.
    pub fn register_operator(
        &self,
        deps: &[NodeId],
        op: OperatorOp,
        opts: OperatorOpts,
    ) -> Result<NodeId, RegisterError> {
        self.register(NodeRegistration {
            deps: deps.to_vec(),
            fn_or_op: Some(NodeFnOrOp::Op(op)),
            opts: NodeOpts {
                equals: opts.equals,
                partial: opts.partial,
                ..NodeOpts::default()
            },
        })
    }

    // -------------------------------------------------------------------
    // Subscription
    // -------------------------------------------------------------------

    /// Subscribe a sink to a node. Returns a [`Subscription`] handle —
    /// dropping the handle unsubscribes the sink. Per §10.12, no manual
    /// `unsubscribe(node, id)` call is required.
    ///
    /// Push-on-subscribe (R1.2.3, R2.2.3 step 4): the sink is registered AFTER
    /// the START handshake fires. The handshake contents depend on node
    /// state:
    /// - Sentinel cache + live (non-terminal): `[START]`
    /// - Cached + live: `[START, DATA(handle)]`
    /// - Cached + terminated (non-resubscribable): `[START, DATA(handle), <terminal>]`
    /// - Sentinel + terminated (non-resubscribable): `[START, <terminal>]`
    ///
    /// Resubscribable terminal lifecycle (R2.2.7 / R2.5.3): if the node was
    /// marked resubscribable via [`Self::set_resubscribable`] AND has
    /// terminated, the subscribe call first **resets** the node — clears
    /// `terminal`, `has_fired_once`, `has_received_teardown`, all
    /// `dep_handles` to `NO_HANDLE`, all `dep_terminals` to `None`, and
    /// drains the pause lockset. The new subscriber then receives a fresh
    /// `[START]` (cache may survive for state nodes; sentinel for compute).
    ///
    /// Activation (R2.2.3 step 5): if this is the first subscriber and the
    /// node is a derived/dynamic compute, recursively activate deps so their
    /// cached handles fill our `dep_handles`.
    #[allow(clippy::needless_pass_by_value)] // Sink is `Arc<dyn Fn>`; we clone for the subscribers map and call it directly. Taking by value matches the ergonomics callers expect.
    pub fn subscribe(&self, node_id: NodeId, sink: Sink) -> Subscription {
        // Subscribe protocol (Slice E rework, post-handshake-reentry-lift):
        //
        // 1. Acquire `wave_owner` first (re-entrant; same-thread passes
        //    through, cross-thread blocks). This is the cross-thread
        //    serialization point that preserves R1.3.5.a happens-after
        //    ordering across the lock-released handshake fire.
        // 2. Acquire state lock briefly: alloc sub_id, run resubscribable
        //    reset if applicable, snapshot handshake state, install sink
        //    in `subscribers`. Drop state lock.
        // 3. Fire handshake LOCK-RELEASED. Per-tier slices (R1.3.5.a):
        //    `[Start]` / `[Data(cache)]?` / `[Complete]?` / `[Error(h)]?`
        //    / `[Teardown]?`. Empty tiers are skipped. Sink callbacks
        //    may re-enter Core freely — same-thread re-entry passes
        //    through `wave_owner` reentrantly.
        // 4. Run activation under `run_wave` if needed (first subscriber
        //    on a non-state node).
        // 5. Drop `wave_owner`.
        //
        // Race-fix discipline: the sink is installed in `subscribers`
        // BEFORE the state lock drops, so concurrent threads that
        // acquire `wave_owner` after our scope sees the sink already
        // registered. Cross-thread emits block on `wave_owner` until
        // we drop it, ensuring all our handshake calls land before
        // any concurrent wave's flush observes the sink.

        // Acquire wave_owner first — cross-thread serialization point.
        // `lock_arc()` is `!Send`; same-thread reentrant.
        let _wave_guard = self.wave_owner.lock_arc();

        let (sub_id, tier_slices, needs_activation, did_reset) = {
            let mut s = self.lock_state();
            let sub_id = s.alloc_sub_id();

            // Resubscribable reset: terminal + flagged → clear lifecycle
            // state so the incoming subscriber starts fresh. F3 audit
            // guard: a node that has received TEARDOWN (R2.6.4) is
            // permanently destroyed at this layer; resurrecting it via a
            // late subscribe is a category error. COMPLETE/ERROR is
            // recoverable for resubscribable nodes; TEARDOWN is not. The
            // handshake will still replay the terminal in the non-reset
            // branch so the late subscriber sees a clean
            // `[START, ?DATA, COMPLETE|ERROR, TEARDOWN]` stream.
            let needs_reset = {
                let rec = s.require_node(node_id);
                rec.resubscribable && rec.terminal.is_some() && !rec.has_received_teardown
            };
            if needs_reset {
                self.reset_for_fresh_lifecycle(&mut s, node_id);
            }

            // Snapshot handshake state under lock.
            let (cache, is_state, first_subscriber, terminal, torn_down) = {
                let rec = s.require_node(node_id);
                (
                    rec.cache,
                    rec.is_state(),
                    rec.subscribers.is_empty(),
                    rec.terminal,
                    rec.has_received_teardown,
                )
            };

            // Build per-tier handshake slices. Each non-empty slice is
            // fired as a separate sink call (R1.3.5.a tier-split).
            let mut tier_slices: SmallVec<[Vec<Message>; 4]> = SmallVec::new();
            tier_slices.push(vec![Message::Start]);
            if cache != NO_HANDLE {
                tier_slices.push(vec![Message::Data(cache)]);
            }
            // Slice E1 (R2.6.5 / Lock 6.G): replay buffered DATA between
            // [Start] (and the cache slice, if present) and any terminal.
            // Each buffered handle becomes a separate per-tier slice so
            // late subscribers see the historical Data sequence as
            // distinct sink calls.
            //
            // Dedupe: when a cache slice is present and the buffer's last
            // entry is the same handle (the typical case — cache always
            // tracks the last DATA emitted, and the buffer's tail entry
            // is that same DATA), skip the last buffer entry to avoid
            // delivering Data(cache) twice. For state nodes whose cache
            // survives unsubscribe, the buffer may have older entries
            // the cache doesn't reflect; the dedupe only drops the
            // single trailing entry that equals cache. (QA A1, 2026-05-07)
            let replay_handles: Vec<HandleId> = {
                let rec = s.require_node(node_id);
                let cap = rec.replay_buffer_cap.unwrap_or(0);
                if cap == 0 {
                    Vec::new()
                } else {
                    let mut v: Vec<HandleId> = rec.replay_buffer.iter().copied().collect();
                    if cache != NO_HANDLE && v.last() == Some(&cache) {
                        v.pop();
                    }
                    v
                }
            };
            for h in &replay_handles {
                tier_slices.push(vec![Message::Data(*h)]);
            }
            if let Some(t) = terminal {
                tier_slices.push(vec![match t {
                    TerminalKind::Complete => Message::Complete,
                    TerminalKind::Error(h) => Message::Error(h),
                }]);
            }
            if torn_down {
                tier_slices.push(vec![Message::Teardown]);
            }

            // Install sink BEFORE dropping state lock so any thread that
            // subsequently acquires `wave_owner` (after our scope ends)
            // sees the sink already registered.
            //
            // Slice X4 / D2: bump `subscribers_revision` alongside the
            // insert so a pending_notify entry opened earlier in the same
            // wave (e.g. inside `batch(|| { emit(s, h1); subscribe(s,
            // late); emit(s, h2); })`) starts a fresh `PendingBatch` on
            // its next `queue_notify` push — making the new sink visible
            // to subsequent emits' flush slices, while the pre-subscribe
            // batch's snapshot stays frozen so we don't double-deliver
            // earlier emits via the wave's flush AND the new sub's
            // handshake replay.
            {
                let rec = s.require_node_mut(node_id);
                rec.subscribers.insert(sub_id, sink.clone());
                rec.subscribers_revision = rec.subscribers_revision.wrapping_add(1);
            }

            let needs_activation = first_subscriber && !is_state;
            (sub_id, tier_slices, needs_activation, needs_reset)
            // state lock drops here
        };

        // Slice E2 (R2.4.6 / D055): on resubscribable terminal reset, fire
        // `wipe_ctx` LOCK-RELEASED so the binding drops its `NodeCtxState`
        // entry (clearing both `store` and any residual `current_cleanup`).
        // The new subscriber's first invoke_fn sees a fresh empty store.
        // Fires AFTER the state lock drops so the binding's
        // `node_ctx.lock()` can't deadlock against Core's state lock — and
        // BEFORE the handshake so the wipe is observable before any
        // user-visible interaction with the new lifecycle.
        if did_reset {
            self.binding.wipe_ctx(node_id);
        }

        // Fire handshake LOCK-RELEASED. Sink may re-enter Core; same-
        // thread re-entry passes through `wave_owner` reentrantly.
        // Cross-thread emits block at `wave_owner` until our scope ends.
        //
        // A7 (Slice F, 2026-05-07): per-tier slice fire is wrapped in
        // `catch_unwind`. The sink is installed in `subscribers` BEFORE
        // the handshake fires (load-bearing — concurrent threads observe
        // the sink immediately). If a sink panics on tier N, the panic
        // would otherwise unwind out of `subscribe` BEFORE the
        // `Subscription` handle is constructed, leaving the sink
        // registered in `subscribers` with no user-held handle to drop.
        // Subsequent waves' `flush_notifications` would re-fire the
        // panicking sink forever.
        //
        // On panic: remove the sink from `subscribers` (via the
        // already-allocated `sub_id`), drop `_wave_guard` cleanly via
        // RAII, and resume the unwind so the user observes the panic at
        // the call site. Same effect as the user dropping the
        // `Subscription` immediately, but pre-emptive.
        for slice in &tier_slices {
            let sink_clone = sink.clone();
            let slice_ref: &[Message] = slice;
            let result = catch_unwind(AssertUnwindSafe(|| sink_clone(slice_ref)));
            if let Err(panic_payload) = result {
                // Remove the orphaned sink. Best-effort: if the node was
                // since torn down (e.g., the sink itself called teardown
                // before panicking), the entry may already be gone.
                {
                    let mut s = self.lock_state();
                    if let Some(rec) = s.nodes.get_mut(&node_id) {
                        rec.subscribers.remove(&sub_id);
                        // Slice X4 / D2: keep revision-tracked snapshot
                        // discipline consistent with the install site —
                        // any pending_notify entry that already absorbed
                        // the panicking sink under the post-install
                        // revision should start a fresh batch on its
                        // next queue_notify push.
                        rec.subscribers_revision = rec.subscribers_revision.wrapping_add(1);
                    }
                }
                std::panic::resume_unwind(panic_payload);
            }
        }

        // Run activation if needed. `run_wave` re-acquires `wave_owner`
        // reentrantly + manages its own state-lock acquisition.
        if needs_activation {
            self.run_wave(|this| {
                let mut s = this.lock_state();
                this.activate_derived(&mut s, node_id);
            });
        }

        Subscription {
            state: Arc::downgrade(&self.state),
            node_id,
            sub_id,
        }
        // _wave_guard drops here, releasing wave_owner.
    }

    /// Mark `node_id` as resubscribable per R2.2.7. Resubscribable nodes
    /// reset their terminal-lifecycle state on a fresh subscribe — see
    /// [`Self::subscribe`].
    ///
    /// Configuration call — must be made before the node has any active
    /// subscribers, since changing the policy mid-flight would surprise
    /// existing observers.
    ///
    /// # Panics
    ///
    /// Panics if the node has subscribers (the policy is observable
    /// behavior; changing it after the fact would change semantics for
    /// existing sinks).
    pub fn set_resubscribable(&self, node_id: NodeId, resubscribable: bool) {
        let mut s = self.lock_state();
        let rec = s.require_node_mut(node_id);
        assert!(
            rec.subscribers.is_empty(),
            "set_resubscribable: node already has subscribers; \
             configure resubscribable before any subscribe call"
        );
        rec.resubscribable = resubscribable;
    }

    /// Reset a resubscribable node's terminal-lifecycle state. Called from
    /// `subscribe` when a late subscriber arrives at a flagged node.
    ///
    /// Released: terminal-slot retain (Error handle), all per-dep terminal
    /// retains (Error handles), all data_batch retains.
    /// Cleared: `terminal`, `has_fired_once`, `has_received_teardown`, all
    /// dep_records to sentinel, the pause lockset (any held locks are
    /// released — replay buffer drops silently because there are no
    /// subscribers to flush to).
    fn reset_for_fresh_lifecycle(&self, s: &mut CoreState, node_id: NodeId) {
        // Phase 1: collect wave-state handle releases + take the old
        // op_scratch + reset other state. Take all mutations under one
        // borrow so the post-borrow phases don't re-walk dep_records.
        let (prev_op, mut old_scratch, handles_to_release, pause_buffer_payloads) = {
            let rec = s.require_node_mut(node_id);
            let mut hs = Vec::new();
            if let Some(TerminalKind::Error(h)) = rec.terminal {
                hs.push(h);
            }
            for dr in &rec.dep_records {
                if let Some(TerminalKind::Error(h)) = dr.terminal {
                    hs.push(h);
                }
                for &h in &dr.data_batch {
                    hs.push(h);
                }
                // Slice C-3 /qa: also release `prev_data`. Prior to this
                // collection, `reset_for_fresh_lifecycle` overwrote
                // `dr.prev_data = NO_HANDLE` without releasing the old
                // handle, leaking one share per dep per resubscribable
                // cycle. The leak was masked because no test exercised
                // the per-dep `prev_data` retain across a lifecycle
                // reset; surfaced by the T1 tightening of
                // `last_releases_buffered_latest_on_lifecycle_reset`.
                if dr.prev_data != NO_HANDLE {
                    hs.push(dr.prev_data);
                }
            }
            // Take pause_state's buffer; collect its payload handles for
            // release (they were retained at queue_notify time; buffer
            // drops because the new subscriber starts fresh).
            let mut pulled = Vec::new();
            if let PauseState::Paused { ref mut buffer, .. } = rec.pause_state {
                for msg in buffer.drain(..) {
                    if let Some(h) = msg.payload_handle() {
                        pulled.push(h);
                    }
                }
            }
            // Slice E1: drain the replay buffer too — the new subscriber
            // gets a fresh lifecycle and shouldn't see prior emissions.
            for h in rec.replay_buffer.drain(..) {
                pulled.push(h);
            }
            // Reset wave / lifecycle state.
            rec.terminal = None;
            rec.has_fired_once = rec.cache != NO_HANDLE && rec.is_state();
            rec.has_received_teardown = false;
            for dr in &mut rec.dep_records {
                dr.prev_data = NO_HANDLE;
                dr.data_batch.clear();
                dr.terminal = None;
                dr.dirty = false;
                dr.involved_this_wave = false;
            }
            rec.pause_state = PauseState::Active;
            rec.involved_this_wave = false;
            rec.dirty = false;
            // P7 (Slice A close /qa): Dynamic nodes clear `tracked` so
            // the post-reset first fire repopulates from the fn's
            // returned tracked-deps set.
            if rec.is_dynamic {
                rec.tracked.clear();
            }
            // Take the old scratch out so we can release its handles and
            // install a fresh one. Operator op is copied for the
            // rebuild step below.
            let prev_op = rec.op;
            let old = std::mem::take(&mut rec.op_scratch);
            (prev_op, old, hs, pulled)
        };

        // Phase 2 (Slice C-3 /qa P1 — RETAIN-BEFORE-RELEASE ordering):
        // build the fresh scratch FIRST, taking new retains on any
        // seed/default handles. This must run BEFORE Phase 3 releases
        // the old scratch's shares — if old `acc` (Scan/Reduce) or old
        // `latest` (Last) aliases the new `seed`/`default` (common:
        // `fold(seed, x) == seed` interns to the same registry entry),
        // releasing the old share first could collapse the binding's
        // registry slot to zero (production bindings remove the value
        // entry on refcount-zero — see `tests/common/mod.rs:191-204`),
        // and a subsequent `retain_handle` on the new seed would bump a
        // refcount on a slot whose value has been removed. By taking
        // the new retains first, we floor the refcount at ≥1 before
        // any release happens.
        let new_scratch = match prev_op {
            // Slice H: the OperatorOp stored on NodeRecord previously
            // passed `make_op_scratch` validation at registration time
            // (RegisterError::OperatorSeedSentinel for Scan/Reduce
            // sentinel seeds; Last { default: NO_HANDLE } is accepted
            // and never errors). Re-running it here on the same op
            // value is structurally guaranteed to succeed.
            Some(op) => self
                .make_op_scratch(op)
                .expect("invariant: stored OperatorOp passed make_op_scratch validation at registration time"),
            None => None,
        };

        // Phase 3: NOW release handles owned by the old op_scratch
        // (Scan/Reduce acc, Distinct/Pairwise prev, Last latest +
        // default). Safe per Phase 2's retain-first floor. The boxed
        // value is consumed and dropped after.
        if let Some(scratch) = old_scratch.as_mut() {
            scratch.release_handles(&*self.binding);
        }
        drop(old_scratch);

        // Phase 4: install the fresh scratch.
        {
            let rec = s.require_node_mut(node_id);
            rec.op_scratch = new_scratch;
        }

        // Phase 5: release wave-state handles collected in phase 1.
        for h in handles_to_release {
            self.binding.release_handle(h);
        }
        for h in pause_buffer_payloads {
            self.binding.release_handle(h);
        }
    }

    /// Activate `root` and any transitive uncached compute deps so their
    /// caches fill our dep_handles slots.
    ///
    /// Slice A close (M1): pure dep-walk + dep_handles population +
    /// pending_fires queueing. No `in_tick` management or `drain_and_flush`
    /// call — the outer caller (typically `Core::subscribe` via
    /// [`Core::run_wave`]) owns the wave lifecycle and drains lock-released
    /// around `invoke_fn`.
    ///
    /// Walk shape:
    ///   1. **Discover phase (DFS via Vec stack):** starting at `root`,
    ///      walk transitively-needing-activation deps via the `deps`
    ///      chain. Build an ordering where each node appears AFTER all
    ///      of its uncached compute deps — i.e., reverse topological
    ///      among the visited subgraph.
    ///   2. **Deliver phase (forward iteration):** for each visited
    ///      node in dep-first order, push deps' caches into the node's
    ///      `dep_handles` slots. Caches that were sentinel pre-walk are
    ///      filled because their parent's fn fires later in the wave's
    ///      drain loop and `commit_emission` propagates new caches forward
    ///      via `deliver_data_to_consumer` — the same path this method
    ///      uses for the initial seed. Adds the node to `pending_fires`
    ///      if its tracked-deps gate is satisfied; the wave-engine drain
    ///      fires the fn lock-released around `invoke_fn`.
    pub(crate) fn activate_derived(&self, s: &mut CoreState, root: NodeId) {
        // Phase 1: discover. DFS to collect every compute node reachable
        // via deps that doesn't yet have a cache and hasn't fired.
        // Record them in dep-first (post-order) so phase 2 can deliver
        // caches forward. Frame is `(node_id, finalize)` — `finalize=false`
        // means "first visit: push deps then re-push self with finalize=true";
        // `finalize=true` means "deps have been expanded, append self to
        // `order`."
        let mut visited: HashSet<NodeId> = HashSet::new();
        let mut order: Vec<NodeId> = Vec::new();
        let mut stack: Vec<(NodeId, bool)> = vec![(root, false)];
        while let Some((id, finalize)) = stack.pop() {
            if finalize {
                order.push(id);
                continue;
            }
            if !visited.insert(id) {
                continue;
            }
            stack.push((id, true));
            let dep_ids: Vec<NodeId> = s.require_node(id).dep_ids_vec();
            for dep_id in dep_ids {
                let (dep_is_state, dep_cache, dep_has_fired) = {
                    let dep_rec = s.require_node(dep_id);
                    (dep_rec.is_state(), dep_rec.cache, dep_rec.has_fired_once)
                };
                if !dep_is_state
                    && dep_cache == NO_HANDLE
                    && !dep_has_fired
                    && !visited.contains(&dep_id)
                {
                    stack.push((dep_id, false));
                }
            }
        }

        // Phase 2: deliver caches in dep-first order. For each node, walk
        // its deps and call `deliver_data_to_consumer` for any with caches.
        // Producer nodes (no deps + has fn — Slice D, D031) have no deps
        // to walk; queue them directly into `pending_fires` so the wave
        // engine fires their fn once on activation.
        for &id in &order {
            let (dep_ids, is_producer) = {
                let rec = s.require_node(id);
                (rec.dep_ids_vec(), rec.is_producer())
            };
            if is_producer {
                s.pending_fires.insert(id);
                continue;
            }
            for (i, dep_id) in dep_ids.iter().copied().enumerate() {
                let dep_cache = s.require_node(dep_id).cache;
                if dep_cache != NO_HANDLE {
                    self.deliver_data_to_consumer(s, id, i, dep_cache);
                }
            }
        }
    }

    // -------------------------------------------------------------------
    // Emission entry point
    // -------------------------------------------------------------------

    /// Set a state node's value. Triggers a wave (DIRTY → DATA/RESOLVED →
    /// fn fires for downstream).
    ///
    /// Silent no-op if the node has already terminated (R1.3.4). The handle
    /// passed in is still released by the caller's binding-side intern path
    /// — no implicit retain is consumed when the call short-circuits.
    ///
    /// # Panics
    ///
    /// Panics if `node_id` is not a state node, or if `new_handle` is
    /// [`NO_HANDLE`] (per R1.2.4, sentinel is not a valid DATA payload).
    pub fn emit(&self, node_id: NodeId, new_handle: HandleId) {
        assert!(
            new_handle != NO_HANDLE,
            "NO_HANDLE is not a valid DATA payload (R1.2.4)"
        );
        // Validate + terminal short-circuit under a brief lock.
        //
        // emit() is valid for State and Producer nodes — both are
        // intrinsic sources whose values are not derived from declared
        // deps. State nodes get emit() from user code; Producer nodes
        // get emit() from sink callbacks the producer's build closure
        // registered (sink fires → re-enter Core → emit on self).
        // Derived / Dynamic / Operator nodes emit via their fn return
        // value through fire_fn / fire_operator, NOT via emit().
        {
            let s = self.lock_state();
            let rec = s.require_node(node_id);
            assert!(
                rec.is_state() || rec.is_producer(),
                "emit() is for state or producer nodes only; \
                 derived/dynamic/operator emit via their fn return value"
            );
            if rec.terminal.is_some() {
                drop(s);
                // Caller's intern share would otherwise leak; cache slot
                // ownership doesn't transfer because we're not advancing
                // cache. Released lock-released so the binding can't
                // deadlock against an internal binding mutex.
                self.binding.release_handle(new_handle);
                return;
            }
        }
        // Run wave — `run_wave` and `commit_emission` manage their own
        // locking; binding callbacks (custom_equals, sinks) fire lock-
        // released.
        self.run_wave(|this| {
            this.commit_emission(node_id, new_handle);
        });
    }

    /// Read a node's current cache. Returns [`NO_HANDLE`] if sentinel.
    #[must_use]
    pub fn cache_of(&self, node_id: NodeId) -> HandleId {
        self.lock_state().require_node(node_id).cache
    }

    /// Whether the node's fn has fired at least once (compute) OR it has had
    /// a non-sentinel value (state).
    #[must_use]
    pub fn has_fired_once(&self, node_id: NodeId) -> bool {
        self.lock_state().require_node(node_id).has_fired_once
    }

    // -------------------------------------------------------------------
    // Read-side inspection helpers (Slice E+, M2)
    //
    // Non-panicking accessors for graph-layer introspection (`describe()`,
    // `observe()`, `node_count()`). All five return Option/empty for
    // unknown ids — they're meant to back walks over `node_ids()` where
    // the caller already knows the ids are valid, plus debugging /
    // dry-run probes that prefer "absence" over "panic".
    //
    // Keep these strictly read-only: no wave entry, no binding callbacks,
    // no lock release. Each takes the state lock once, copies a small
    // value, and drops the lock.
    // -------------------------------------------------------------------

    /// Snapshot of every registered `NodeId` in unspecified order. The
    /// order matches `HashMap` iteration over the internal node table —
    /// callers that need stable ordering should track names at the
    /// `Graph` layer (canonical spec §3.5 namespace).
    #[must_use]
    pub fn node_ids(&self) -> Vec<NodeId> {
        self.lock_state().nodes.keys().copied().collect()
    }

    /// Total number of nodes registered in this Core.
    #[must_use]
    pub fn node_count(&self) -> usize {
        self.lock_state().nodes.len()
    }

    /// Returns `Some(kind)` for known nodes, `None` for unknown. Kind is
    /// **derived** from the field shape per D030 — see [`NodeKind`].
    #[must_use]
    pub fn kind_of(&self, node_id: NodeId) -> Option<NodeKind> {
        self.lock_state().nodes.get(&node_id).map(NodeRecord::kind)
    }

    /// Snapshot of the node's deps in declaration order. Empty for
    /// unknown nodes or for state nodes (which have no deps).
    #[must_use]
    pub fn deps_of(&self, node_id: NodeId) -> Vec<NodeId> {
        self.lock_state()
            .nodes
            .get(&node_id)
            .map(NodeRecord::dep_ids_vec)
            .unwrap_or_default()
    }

    /// Returns `Some(kind)` if the node has terminated (R1.3.4) — the
    /// pair `Some(Complete)` / `Some(Error(h))` mirrors the wire message
    /// the node emitted. `None` for live nodes or unknown ids.
    #[must_use]
    pub fn is_terminal(&self, node_id: NodeId) -> Option<TerminalKind> {
        self.lock_state()
            .nodes
            .get(&node_id)
            .and_then(|r| r.terminal)
    }

    /// Whether the node has wave-scoped DIRTY pending (a tier-1 message
    /// queued but the matching tier-3 settle has not yet flushed).
    /// `false` for unknown ids. Mostly useful for `describe()` status
    /// classification (R3.6.1 `"dirty"`).
    #[must_use]
    pub fn is_dirty(&self, node_id: NodeId) -> bool {
        self.lock_state()
            .nodes
            .get(&node_id)
            .is_some_and(|r| r.dirty)
    }

    /// Snapshot of `parent`'s meta companion list (R1.3.9.d / R2.3.3 —
    /// the companions added via [`Self::add_meta_companion`]). Empty
    /// for unknown ids or for nodes with no companions registered.
    ///
    /// Used by the graph layer's `signal_invalidate` to filter meta
    /// children out of the broadcast (canonical R3.7.2 — meta caches
    /// are preserved across graph-wide INVALIDATE).
    #[must_use]
    pub fn meta_companions_of(&self, parent: NodeId) -> Vec<NodeId> {
        self.lock_state()
            .nodes
            .get(&parent)
            .map(|r| r.meta_companions.clone())
            .unwrap_or_default()
    }

    // -------------------------------------------------------------------
    // Wave engine — lives in `crate::batch` (Slice C-1 module split;
    // Slice A close M1 refactor lifted the binding-callback re-entrance
    // restrictions). The methods are still on `Core`; see `batch.rs` for:
    //
    //   - `run_wave` — wave entry, manages own locking.
    //   - `drain_and_flush` — drain phase, lock-released around invoke_fn.
    //   - `commit_emission` — lock-released around custom_equals.
    //   - `pick_next_fire`, `deliver_data_to_consumer`, `queue_notify`,
    //     `flush_notifications` — wave-engine helpers.
    // -------------------------------------------------------------------
}

// -----------------------------------------------------------------------
// COMPLETE / ERROR — terminal lifecycle + auto-cascade gating
// -----------------------------------------------------------------------

impl Core {
    /// Emit `[COMPLETE]` (R1.3.4) on `node_id`, marking it terminal. After
    /// this call:
    ///
    /// - Subsequent `Core::emit` on this node is a silent no-op (idempotent
    ///   termination).
    /// - The node's fn no longer fires.
    /// - The node's cache is preserved (last value still observable via
    ///   `cache_of`).
    /// - Children receive `[COMPLETE]` (tier 5 — bypasses pause buffer).
    /// - Auto-cascade gating (Lock 2.B): each child that has all of its
    ///   deps in a terminal state auto-emits its own `[COMPLETE]`. ERROR
    ///   dominates COMPLETE — if any of a child's deps emitted ERROR, the
    ///   child auto-cascades that ERROR instead.
    ///
    /// Idempotent: calling `complete` on an already-terminal node is a no-op.
    ///
    /// # Panics
    ///
    /// Panics if `node_id` is unknown.
    pub fn complete(&self, node_id: NodeId) {
        self.emit_terminal(node_id, TerminalKind::Complete);
    }

    /// Emit `[ERROR, error_handle]` (R1.3.4) on `node_id`. `error_handle`
    /// must resolve to a non-sentinel value (R1.2.5) — the binding side has
    /// already interned the error value before this call. Same lifecycle
    /// effects as [`Self::complete`]; ERROR dominates COMPLETE in auto-
    /// cascade gating.
    ///
    /// # Panics
    ///
    /// Panics if `node_id` is unknown or `error_handle == NO_HANDLE`.
    pub fn error(&self, node_id: NodeId, error_handle: HandleId) {
        assert!(
            error_handle != NO_HANDLE,
            "NO_HANDLE is not a valid ERROR payload (R1.2.5)"
        );
        self.emit_terminal(node_id, TerminalKind::Error(error_handle));
        // The caller's intern share for `error_handle` is NOT transferred
        // to any slot — `terminate_node` takes its OWN retain for every
        // populated `terminal` and `dep_terminals` slot. Release the
        // caller's share here (mirrors `Core::emit`'s short-circuit
        // release on terminal). Without this, every `error()` call leaks
        // one binding-side handle ref. Slice A-bigger /qa item D fix.
        self.binding.release_handle(error_handle);
    }

    fn emit_terminal(&self, node_id: NodeId, terminal: TerminalKind) {
        {
            let s = self.lock_state();
            assert!(s.nodes.contains_key(&node_id), "unknown node {node_id:?}");
        }
        // Wave runs with `run_wave` orchestrating drain. The thunk acquires
        // its own lock to queue the cascade (terminate_node is a fast
        // structural walk; no binding callbacks beyond non-re-entrant
        // retain/release).
        self.run_wave(|this| {
            let mut s = this.lock_state();
            this.terminate_node(&mut s, node_id, terminal);
        });
    }

    /// Set the node's terminal slot, queue the wire message, and cascade to
    /// children. Idempotent on already-terminal node (no-op).
    ///
    /// Iterative implementation (Slice A-bigger, M1-close): a work-queue
    /// drives the cascade so deep linear chains don't overflow the OS
    /// thread stack. Mirrors `path_from_to`'s explicit-stack pattern.
    fn terminate_node(&self, s: &mut CoreState, node_id: NodeId, terminal: TerminalKind) {
        let mut work: Vec<(NodeId, TerminalKind)> = vec![(node_id, terminal)];
        while let Some((id, t)) = work.pop() {
            if s.require_node(id).terminal.is_some() {
                continue; // Idempotent — already terminal.
            }
            // Take a refcount share for the terminal slot so the error
            // handle outlives the binding-side intern's transient share.
            if let TerminalKind::Error(h) = t {
                self.binding.retain_handle(h);
            }
            // Slice E2 /qa Q2(b) (D069): if a resubscribable node is
            // terminating with no live subscribers, queue eager
            // `wipe_ctx` for the wave's lock-released drain. This is the
            // mutually-exclusive complement of the `Subscription::Drop`
            // wipe site: when the LAST sub drops first then terminate
            // fires, subs are empty here and we queue; when terminate
            // fires WITH subs still alive, we DON'T queue (subs not
            // empty), and `Subscription::Drop` will fire wipe directly
            // when those subs eventually drop. Either way, exactly one
            // wipe fires per terminal lifecycle.
            let queue_wipe = {
                let rec = s.require_node(id);
                rec.resubscribable && rec.subscribers.is_empty()
            };
            s.require_node_mut(id).terminal = Some(t);
            if queue_wipe {
                s.pending_wipes.push(id);
            }
            // Drain pending fires for this node — fn won't fire on a
            // terminal node.
            s.pending_fires.remove(&id);
            // R1.3.8.b / Slice F (A3, 2026-05-07): if this node was paused
            // when terminating (the canonical case is the R1.3.8.c overflow
            // ERROR synthesis path), drain the pause buffer and release
            // each payload's queue_notify-time retain. Without this, the
            // buffer leaks one share per buffered DATA/RESOLVED/INVALIDATE.
            // Subscribers receive the terminal directly via the cascade
            // below (tier-5 bypasses the pause buffer); the buffered
            // content is moot post-terminal.
            let drained: Vec<HandleId> = {
                let rec = s.require_node_mut(id);
                let mut drained: Vec<HandleId> = Vec::new();
                if rec.pause_state.is_paused() {
                    // Take the buffered messages out, then collapse the
                    // pause state to Active so subsequent code observes a
                    // clean lifecycle. Idempotent on Active (no-op).
                    let prev = std::mem::replace(&mut rec.pause_state, PauseState::Active);
                    if let PauseState::Paused { buffer, .. } = prev {
                        drained.extend(buffer.into_iter().filter_map(Message::payload_handle));
                    }
                }
                // QA A4 (2026-05-07): drain replay buffer on terminate. A
                // non-resubscribable terminal ends the lifecycle; without
                // this drain the buffer's retains leak until `Drop for
                // CoreState`. Resubscribable nodes' replay buffers are
                // also drained (when they're hit by a terminal cascade);
                // a fresh subscribe rebuilds the buffer from scratch as
                // part of `reset_for_fresh_lifecycle`.
                drained.extend(rec.replay_buffer.drain(..));
                drained
            };
            for h in drained {
                self.binding.release_handle(h);
            }
            // Queue the wire message (tier 5 — bypasses pause buffer).
            let msg = match t {
                TerminalKind::Complete => Message::Complete,
                TerminalKind::Error(h) => Message::Error(h),
            };
            self.queue_notify(s, id, msg);
            // Cascade to children.
            let child_ids: Vec<NodeId> = s
                .children
                .get(&id)
                .map(|c| c.iter().copied().collect())
                .unwrap_or_default();
            for child_id in child_ids {
                let dep_idx = s.require_node(child_id).dep_index_of(id);
                let Some(idx) = dep_idx else { continue };
                // Mark this child's per-dep terminal slot. Take a retain on
                // the error handle for the slot share.
                {
                    let child = s.require_node_mut(child_id);
                    if child.dep_records[idx].terminal.is_some() {
                        // Idempotent — child already saw this dep terminate.
                        continue;
                    }
                    child.dep_records[idx].terminal = Some(t);
                }
                if let TerminalKind::Error(h) = t {
                    self.binding.retain_handle(h);
                }
                // Auto-cascade gating: if all deps now terminal, push child
                // onto the work queue with the chosen terminal.
                //
                // Slice C-1: kinds that opt out of Lock 2.B (currently
                // `Operator(Reduce)`) intercept upstream COMPLETE so they
                // can emit their accumulator before terminating. Instead of
                // cascading, queue the child for fn-fire — `fire_operator`
                // sees `dep_records[0].terminal` set and emits the
                // appropriate batch (Data(acc) + Complete on COMPLETE,
                // Error(h) on ERROR).
                let action = {
                    let child = s.require_node(child_id);
                    if child.terminal.is_some() {
                        ChildAction::None // Already terminated — no-op.
                    } else if child.all_deps_terminal() {
                        if child.skips_auto_cascade() {
                            ChildAction::QueueFire
                        } else {
                            ChildAction::Cascade(pick_cascade_terminal(&child.dep_records))
                        }
                    } else {
                        ChildAction::None
                    }
                };
                match action {
                    ChildAction::None => {}
                    ChildAction::Cascade(t_child) => {
                        work.push((child_id, t_child));
                    }
                    ChildAction::QueueFire => {
                        s.pending_fires.insert(child_id);
                    }
                }
            }
        }
    }
}

/// Outcome of Lock 2.B child gating in `terminate_node`'s cascade walk.
enum ChildAction {
    /// No cascade; child is already terminal or not yet all-deps-terminal.
    None,
    /// Auto-cascade with the picked terminal kind (ERROR dominates COMPLETE).
    Cascade(TerminalKind),
    /// Queue child for fn-fire instead of cascading. Used by operator
    /// kinds that intercept upstream terminal (Operator(Reduce)).
    QueueFire,
}

/// Lock 2.B cascade-terminal selection: ERROR dominates COMPLETE; first
/// ERROR seen wins. Caller has already verified all deps are terminal.
fn pick_cascade_terminal(dep_records: &[DepRecord]) -> TerminalKind {
    for dr in dep_records {
        if let Some(TerminalKind::Error(h)) = dr.terminal {
            return TerminalKind::Error(h);
        }
    }
    TerminalKind::Complete
}

// -----------------------------------------------------------------------
// TEARDOWN — destruction, with auto-COMPLETE prepend (R2.6.4 / Lock 6.F)
// -----------------------------------------------------------------------

impl Core {
    /// Tear `node_id` down. Per R2.6.4 / Lock 6.F:
    ///
    /// - **Auto-prepend COMPLETE.** If the node has not yet emitted a
    ///   terminal (`COMPLETE` / `ERROR`), `terminate_node` is called with
    ///   `Complete` first so subscribers see `[COMPLETE, TEARDOWN]`, not
    ///   bare `[TEARDOWN]`. This guarantees a clean end-of-stream signal
    ///   to async iterators and other consumers that wait on terminal
    ///   delivery.
    /// - **Idempotent on duplicate delivery.** The per-node
    ///   `has_received_teardown` flag is set on the first call; subsequent
    ///   `teardown` calls (or cascade visits from other paths) are silent
    ///   no-ops — no second `[COMPLETE, TEARDOWN]` pair to subscribers.
    /// - **Cascade downstream.** Each child is recursively torn down. The
    ///   child's own COMPLETE auto-cascades from `terminate_node`'s logic
    ///   (Lock 2.B); its TEARDOWN comes from this cascade.
    ///
    /// # Panics
    ///
    /// Panics if `node_id` is unknown.
    pub fn teardown(&self, node_id: NodeId) {
        {
            let s = self.lock_state();
            assert!(s.nodes.contains_key(&node_id), "unknown node {node_id:?}");
        }
        let torn_down: Arc<Mutex<Vec<NodeId>>> = Arc::new(Mutex::new(Vec::new()));
        let torn_down_for_wave = torn_down.clone();
        self.run_wave(move |this| {
            let mut s = this.lock_state();
            let collected = this.teardown_inner(&mut s, node_id);
            torn_down_for_wave.lock().extend(collected);
        });
        // Fire NodeTornDown for every cascaded id (root + metas +
        // downstream consumers that auto-cascaded). Outside the state
        // lock, matching fire_topology_event discipline.
        let ids = std::mem::take(&mut *torn_down.lock());
        for id in ids {
            self.fire_topology_event(&crate::topology::TopologyEvent::NodeTornDown(id));
        }
    }

    /// Iterative teardown walk (Slice A-bigger, M1-close).
    ///
    /// The recursive shape was:
    ///   ```text
    ///   teardown(n):
    ///     if torn_down: return
    ///     mark torn_down
    ///     for meta in metas: teardown(meta)
    ///     terminate_node + queue Teardown
    ///     for child in children: teardown(child)
    ///   ```
    /// Deep linear chains (~10k nodes) overflowed the OS thread stack.
    ///
    /// The iterative shape uses a `Vec<Action>` stack with `Visit` and
    /// `EmitTeardown` actions. `Visit(n)` marks `n` torn-down (or no-ops
    /// if already), then pushes (in reverse order so LIFO pops in forward
    /// order) `Visit(child_K), …, Visit(child_1), EmitTeardown(n),
    /// Visit(meta_M), …, Visit(meta_1)`. The R1.3.9.d "metas first, then
    /// self, then children" ordering is preserved by the push order:
    /// metas pop first, recursively expand and emit; then `EmitTeardown(n)`
    /// pops and runs `terminate_node` + queue `Teardown`; then children
    /// pop. Idempotency via `has_received_teardown` keeps each node
    /// visited at most once even when multi-parent diamonds re-enter via
    /// a sibling path.
    fn teardown_inner(&self, s: &mut CoreState, root: NodeId) -> Vec<NodeId> {
        enum Action {
            Visit(NodeId),
            EmitTeardown(NodeId),
        }
        let mut stack: Vec<Action> = vec![Action::Visit(root)];
        // Topology accumulator: every node that actually emits TEARDOWN
        // (i.e. each `EmitTeardown(id)` site, NOT each `Visit` — visits
        // for already-torn-down nodes short-circuit on idempotency).
        let mut torn_down: Vec<NodeId> = Vec::new();
        while let Some(action) = stack.pop() {
            match action {
                Action::Visit(id) => {
                    if s.require_node(id).has_received_teardown {
                        continue; // Idempotent (R2.6.4).
                    }
                    s.require_node_mut(id).has_received_teardown = true;
                    // Push order: children first (pop LAST), then
                    // EmitTeardown(id), then metas (pop FIRST). Reverse
                    // each list so within-group order matches the original
                    // recursive iteration.
                    let children: Vec<NodeId> = s
                        .children
                        .get(&id)
                        .map(|c| c.iter().copied().collect())
                        .unwrap_or_default();
                    for &child in children.iter().rev() {
                        stack.push(Action::Visit(child));
                    }
                    stack.push(Action::EmitTeardown(id));
                    let metas: Vec<NodeId> = s.require_node(id).meta_companions.clone();
                    for &meta in metas.iter().rev() {
                        stack.push(Action::Visit(meta));
                    }
                }
                Action::EmitTeardown(id) => {
                    // Auto-prepend COMPLETE if not yet terminal. The (now
                    // iterative) terminate_node handles auto-cascade to
                    // children's own terminal slots per Lock 2.B.
                    let already_terminal = s.require_node(id).terminal.is_some();
                    if !already_terminal {
                        self.terminate_node(s, id, TerminalKind::Complete);
                    }
                    // Wire emission of the TEARDOWN itself (tier 6).
                    self.queue_notify(s, id, Message::Teardown);
                    torn_down.push(id);
                }
            }
        }
        torn_down
    }

    /// Attach `companion` as a meta companion of `parent` per R1.3.9.d.
    /// Meta companions are nodes whose lifecycle is bound to the parent's
    /// in TEARDOWN ordering: when `parent` tears down, `companion` tears
    /// down first.
    ///
    /// Use this for inspection / audit / sidecar nodes that subscribe to
    /// parent state — without the ordering, the companion could observe
    /// the parent mid-destruction and emit garbage.
    ///
    /// Idempotent on duplicate registration of the same companion.
    ///
    /// # Lifecycle constraint
    ///
    /// Intended for **setup-time** wiring — call this before `parent` or
    /// `companion` enters a wave. Mid-wave registration (especially during
    /// a teardown cascade in flight) is implementation-defined: the new
    /// edge takes effect on the *next* wave. Adding a companion to a
    /// torn-down parent silently no-ops (the parent will not tear down
    /// again). For dynamic companion attachment with deterministic
    /// ordering, prefer constructing the wiring before subscribers exist.
    ///
    /// # Panics
    ///
    /// Panics if either node id is unknown, or if `parent == companion`
    /// (a node cannot be its own meta companion — would loop on TEARDOWN).
    pub fn add_meta_companion(&self, parent: NodeId, companion: NodeId) {
        assert!(parent != companion, "node cannot be its own meta companion");
        let mut s = self.lock_state();
        assert!(s.nodes.contains_key(&parent), "unknown parent {parent:?}");
        assert!(
            s.nodes.contains_key(&companion),
            "unknown companion {companion:?}"
        );
        let metas = &mut s.require_node_mut(parent).meta_companions;
        if !metas.contains(&companion) {
            metas.push(companion);
        }
    }
}

// -----------------------------------------------------------------------
// INVALIDATE — cache clear + downstream cascade
// -----------------------------------------------------------------------

impl Core {
    /// Clear `node_id`'s cache and cascade `[INVALIDATE]` to downstream
    /// dependents per canonical spec §1.4.
    ///
    /// Semantics:
    /// - **Never-populated case (R1.4 line 197):** if `cache == NO_HANDLE`,
    ///   the call is a no-op — no cache to clear, no INVALIDATE emitted.
    ///   This naturally provides idempotency within a wave: once a node has
    ///   been invalidated this wave (cache = NO_HANDLE), a second invalidate
    ///   on the same node does nothing.
    /// - **Cache clear (immediate):** the node's cached handle is dropped
    ///   (refcount released), `cache` becomes `NO_HANDLE`. State nodes
    ///   keep `has_fired_once` per spec — INVALIDATE is not a re-gating
    ///   event (the next emission to a previously-fired state still does
    ///   not re-trigger the first-run gate; that's a resubscribable-terminal
    ///   lifecycle concern, separate slice).
    /// - **Wire emission (tier 4):** `[INVALIDATE]` is queued via the
    ///   normal pause-aware notify path. Buffers while paused, flushes
    ///   immediately otherwise.
    /// - **Downstream cascade:** for each child of this node, the child's
    ///   `dep_handles[idx_of_node]` is reset to `NO_HANDLE` (its previous
    ///   value referenced a now-released handle). The child is then
    ///   recursively invalidated (no-op if its cache was already
    ///   `NO_HANDLE`). This re-closes the child's first-run gate — fn
    ///   won't fire again until the upstream re-emits a value.
    ///
    /// Wraps in a fresh wave when called from outside a wave, so
    /// notifications flush at the natural wave boundary.
    ///
    /// # Panics
    ///
    /// Panics if `node_id` is unknown, consistent with `emit` / `pause`.
    pub fn invalidate(&self, node_id: NodeId) {
        {
            let s = self.lock_state();
            assert!(s.nodes.contains_key(&node_id), "unknown node {node_id:?}");
        }
        self.run_wave(|this| {
            let mut s = this.lock_state();
            this.invalidate_inner(&mut s, node_id);
        });
    }

    /// Iterative invalidate cascade (Slice A-bigger, M1-close).
    ///
    /// The recursive shape was a depth-first cache-clear walk:
    ///   ```text
    ///   invalidate(n):
    ///     if cache(n) == NO_HANDLE: return  // already-invalidated guard
    ///     cache(n) = NO_HANDLE; release handle
    ///     queue Invalidate(n)
    ///     for child in children:
    ///       child.dep_handles[idx] = NO_HANDLE
    ///       invalidate(child)
    ///   ```
    /// Deep linear chains overflowed the OS thread stack. The work-queue
    /// rewrite has no ordering subtleties (unlike teardown's R1.3.9.d
    /// metas-first constraint) — Invalidate is a tier-4 broadcast where
    /// the never-populated / already-invalidated guard provides natural
    /// idempotency for diamond fan-in.
    fn invalidate_inner(&self, s: &mut CoreState, root: NodeId) {
        let mut work: Vec<NodeId> = vec![root];
        while let Some(node_id) = work.pop() {
            // Never-populated / already-invalidated: no-op (R1.4 idempotency).
            // Per R1.3.9.c never-populated case, OnInvalidate cleanup hook
            // also does NOT fire — natural fallout of skipping via the
            // cache==NO_HANDLE guard (we never reach the queue-push below).
            let old_handle = s.require_node(node_id).cache;
            if old_handle == NO_HANDLE {
                continue;
            }
            // Clear cache + release the handle's slot ownership.
            s.require_node_mut(node_id).cache = NO_HANDLE;
            self.binding.release_handle(old_handle);
            // Slice E2 (R1.3.9.b strict per D057 + D058 fire-at-cache-clear):
            // queue OnInvalidate cleanup hook for lock-released drain at
            // wave-end. The dedup set guarantees at-most-once-per-wave-per-
            // node firing even if a node re-populates mid-wave (via fn-fire
            // emit) and gets re-invalidated through a separate path. Pure
            // cache==NO_HANDLE idempotency (above) catches "still at
            // sentinel" only; the explicit set is the strict R1.3.9.b
            // reading.
            if s.invalidate_hooks_fired_this_wave.insert(node_id) {
                s.deferred_cleanup_hooks
                    .push((node_id, CleanupTrigger::OnInvalidate));
            }
            // Wire emission. Pause-aware via queue_notify.
            self.queue_notify(s, node_id, Message::Invalidate);
            // Cascade: for each child, clear the dep record's prev_data
            // referencing this node and push child onto the work queue.
            let child_ids: Vec<NodeId> = s
                .children
                .get(&node_id)
                .map(|c| c.iter().copied().collect())
                .unwrap_or_default();
            for child_id in child_ids {
                let dep_idx = s.require_node(child_id).dep_index_of(node_id);
                if let Some(idx) = dep_idx {
                    // Reset the child's dep record — the handle was just
                    // released. Subsequent first-run-gate checks see
                    // sentinel and re-close.
                    //
                    // Snapshot prev_data + data_batch retains for deferred
                    // release, then clear the record. Two-phase to satisfy
                    // the borrow checker (nodes + deferred_handle_releases
                    // are separate CoreState fields).
                    let (old_prev, batch_hs): (HandleId, SmallVec<[HandleId; 1]>) = {
                        let dr = &s.require_node(child_id).dep_records[idx];
                        (dr.prev_data, dr.data_batch.clone())
                    };
                    if old_prev != NO_HANDLE {
                        s.deferred_handle_releases.push(old_prev);
                    }
                    for h in batch_hs {
                        s.deferred_handle_releases.push(h);
                    }
                    let dr = &mut s.require_node_mut(child_id).dep_records[idx];
                    dr.prev_data = NO_HANDLE;
                    dr.data_batch.clear();
                    work.push(child_id);
                }
            }
        }
    }
}

// -----------------------------------------------------------------------
// PAUSE / RESUME — multi-pauser lockset + replay buffer
// -----------------------------------------------------------------------

/// Reported back from [`Core::resume`] when the final lock releases.
///
/// `replayed` is the number of tier-3/tier-4 messages dispatched to
/// subscribers as part of the drain. `dropped` is the number of messages
/// that fell out the front of the buffer due to the Core-global
/// `pause_buffer_cap` while this pause cycle was active. A non-zero
/// `dropped` indicates a controller held the lock long enough to overflow
/// the cap; the binding may want to surface a warning or error.
#[derive(Copy, Clone, Debug, PartialEq, Eq)]
pub struct ResumeReport {
    pub replayed: u32,
    pub dropped: u32,
}

impl Core {
    /// Acquire a pause lock on `node_id`. The first lock transitions the
    /// node from `Active` to `Paused`; further locks add to the lockset.
    /// While paused, tier-3 (DATA/RESOLVED) and tier-4 (INVALIDATE) outgoing
    /// messages buffer in the node's pause buffer; other tiers flush
    /// immediately.
    ///
    /// Re-acquiring the same `lock_id` is an idempotent no-op (matches TS
    /// convention, R1.2.6 silent on the case).
    pub fn pause(&self, node_id: NodeId, lock_id: LockId) -> Result<(), PauseError> {
        let mut s = self.lock_state();
        let rec = s
            .nodes
            .get_mut(&node_id)
            .ok_or(PauseError::UnknownNode(node_id))?;
        // QA A5 (2026-05-07): terminated nodes can't be re-paused. Without
        // this check, a stale pause-controller calling pause() on an
        // already-terminated node would re-arm `pause_state` to Paused.
        // The terminate_node path collapses pause_state → Active and
        // drains the buffer (A3-related), but doesn't gate subsequent
        // pause() calls. Treat as idempotent no-op (consistent with how
        // emit/complete/error early-return on terminal).
        if rec.terminal.is_some() {
            return Ok(());
        }
        // Slice F audit close (2026-05-07): `PausableMode::Off` means the
        // dispatcher ignores PAUSE for this node — tier-3 flushes
        // immediately, fn fires immediately. Treat the call as a successful
        // no-op so callers don't need to special-case.
        if rec.pausable == PausableMode::Off {
            return Ok(());
        }
        rec.pause_state.add_lock(lock_id);
        Ok(())
    }

    /// Release a pause lock on `node_id`. If the lockset becomes empty, the
    /// node transitions back to `Active` and the buffered messages are
    /// dispatched to subscribers in arrival order. Returns a [`ResumeReport`]
    /// when the final lock released; `None` if the lockset is still
    /// non-empty (further locks held).
    ///
    /// Releasing an unknown `lock_id` (or releasing on an already-Active
    /// node) is an idempotent no-op returning `None`.
    pub fn resume(
        &self,
        node_id: NodeId,
        lock_id: LockId,
    ) -> Result<Option<ResumeReport>, PauseError> {
        // Phase 1 (lock-held): collect drained buffer + pending-wave flag +
        // sink Arcs. For default-mode nodes whose `pending_wave` was set
        // during pause, schedule a single fn-fire by adding to
        // `pending_fires` BEFORE we exit the lock — the wave engine picks
        // it up on the next drain tick.
        let (sinks, messages, dropped, pending_wave_for_default) = {
            let mut s = self.lock_state();
            let rec = s
                .nodes
                .get_mut(&node_id)
                .ok_or(PauseError::UnknownNode(node_id))?;
            // For Off mode, pause/resume are no-ops by construction.
            if rec.pausable == PausableMode::Off {
                return Ok(None);
            }
            let was_default_mode = rec.pausable == PausableMode::Default;
            // Capture pending_wave BEFORE remove_lock collapses the state.
            let pending_wave = if was_default_mode {
                rec.pause_state.take_pending_wave()
            } else {
                false
            };
            let Some((buffer, dropped)) = rec.pause_state.remove_lock(lock_id) else {
                // Not the final-resume — restore the pending_wave flag we
                // tentatively cleared, since we're not transitioning to
                // Active yet.
                if pending_wave {
                    rec.pause_state.mark_pending_wave();
                }
                return Ok(None);
            };
            let sinks: Vec<Sink> = rec.subscribers.values().cloned().collect();
            let messages: Vec<Message> = buffer.into_iter().collect();
            // Default-mode pending-wave handling: schedule the fn-fire so
            // the wave engine consolidates the pause-window dep deliveries
            // into one fn execution. State nodes don't fire fn (no
            // `pending_fires` membership has effect for them).
            if pending_wave && was_default_mode {
                s.pending_fires.insert(node_id);
            }
            (sinks, messages, dropped, pending_wave && was_default_mode)
        };
        let replayed = u32::try_from(messages.len()).unwrap_or(u32::MAX);

        // Phase 2 (lock-released): fire sinks for ResumeAll-buffered
        // messages. Default-mode resume produces no buffered replay (the
        // consolidated fn-fire produces fresh wave traffic via the standard
        // commit_emission path).
        if !messages.is_empty() {
            for sink in &sinks {
                sink(&messages);
            }
            // Phase 3: balance the retain_handle calls done at buffer-push
            // time — sinks observe values but don't own refcount shares.
            for msg in &messages {
                if let Some(h) = msg.payload_handle() {
                    self.binding.release_handle(h);
                }
            }
        }

        // Phase 4 (default-mode): drain the consolidated fn-fire scheduled
        // in Phase 1. `run_wave` re-acquires `wave_owner` reentrantly + runs
        // the standard drain pipeline; the new fn-fire emerges as a normal
        // wave's worth of messages to subscribers.
        if pending_wave_for_default {
            self.run_wave(|_this| {
                // The pending_fires entry was pushed in Phase 1 under the
                // lock. run_wave's drain picks it up.
            });
        }
        Ok(Some(ResumeReport { replayed, dropped }))
    }

    /// True if the node currently holds at least one pause lock.
    #[must_use]
    pub fn is_paused(&self, node_id: NodeId) -> bool {
        self.state
            .lock()
            .require_node(node_id)
            .pause_state
            .is_paused()
    }

    /// Number of pause locks currently held on `node_id`. `0` if Active.
    #[must_use]
    pub fn pause_lock_count(&self, node_id: NodeId) -> usize {
        self.state
            .lock()
            .require_node(node_id)
            .pause_state
            .lock_count()
    }

    /// Test helper: whether `node_id` currently holds the given `lock_id`.
    #[must_use]
    pub fn holds_pause_lock(&self, node_id: NodeId, lock_id: LockId) -> bool {
        self.state
            .lock()
            .require_node(node_id)
            .pause_state
            .contains_lock(lock_id)
    }
}

// -----------------------------------------------------------------------
// set_deps — atomic dep mutation
// -----------------------------------------------------------------------

/// Errors returnable by [`Core::set_deps`].
///
/// Per `~/src/graphrefly-ts/docs/research/rewire-design-notes.md` and the
/// Phase 13.8 Q1 lock:
/// - `SelfDependency` — `n in newDeps` (self-loops are pathological without
///   explicit fixed-point semantics, which GraphReFly does not provide).
/// - `WouldCreateCycle { path }` — adding the new edge would create a cycle.
///   The `path` field reports the offending dep chain for debuggability.
/// - `UnknownNode` / `NotComputeNode` — invariant violations from the caller.
/// - `TerminalNode` — `n` itself has emitted COMPLETE/ERROR; rewiring a
///   terminal stream is a category error (terminal is one-shot at this
///   layer; recovery is the resubscribable path on a fresh subscribe).
/// - `TerminalDep` — a newly-added dep is terminal AND not resubscribable.
///   Resubscribable terminal deps are accepted because the subscribe path
///   resets their lifecycle. Non-resubscribable terminal deps would deliver
///   their already-emitted terminal directly to `n`'s `dep_terminals` slot,
///   which is rarely intended.
#[derive(Error, Debug, Clone, PartialEq)]
pub enum SetDepsError {
    /// `n` appeared in `new_deps` (self-loop rejection).
    #[error("set_deps({n:?}, ...): self-dependency rejected (n appeared in new_deps)")]
    SelfDependency { n: NodeId },

    /// Adding the new dep would create a cycle. `path` is the chain
    /// `[added_dep, ..., n]` reachable via existing deps.
    #[error(
        "set_deps({n:?}, ...): cycle would form via path {path:?} \
         (adding {added_dep:?} → {n:?} closes the loop)"
    )]
    WouldCreateCycle {
        n: NodeId,
        added_dep: NodeId,
        path: Vec<NodeId>,
    },

    #[error("set_deps: unknown node {0:?}")]
    UnknownNode(NodeId),

    #[error("set_deps: node {0:?} is not a compute node (state nodes have no deps)")]
    NotComputeNode(NodeId),

    /// `n` itself has terminated (COMPLETE / ERROR). Rewiring a terminal node
    /// is rejected — the stream has ended at this layer. To recover, mark
    /// the node resubscribable before terminate; a fresh subscribe will then
    /// reset its lifecycle.
    #[error("set_deps({n:?}, ...): node has already terminated; cannot rewire a terminal node")]
    TerminalNode { n: NodeId },

    /// A newly-added dep is terminal AND non-resubscribable. Per Phase 13.8
    /// Q1, this is rejected; resubscribable terminal deps are allowed
    /// because the subscribe path resets them when activated. Already-present
    /// terminal deps are unaffected (their terminal status was accepted at
    /// the time they terminated).
    #[error(
        "set_deps({n:?}, ...): added dep {dep:?} is terminal and not resubscribable; \
         either mark it resubscribable before terminate, or remove the dep from new_deps"
    )]
    TerminalDep { n: NodeId, dep: NodeId },

    /// `n` itself is currently mid-fire — a user fn for `n` re-entered Core
    /// via `set_deps(n, ...)` from inside `n`'s own `invoke_fn` /
    /// `project_each` / `predicate_each` / etc. Phase 1 of the dispatcher
    /// snapshotted `dep_handles` BEFORE the lock-released callback; the
    /// callback returning a `tracked` set indexed against THAT ordering
    /// would corrupt indices if the rewire re-orders deps mid-fire.
    /// Rejected to preserve the dynamic-tracked-indices invariant (D1).
    ///
    /// Workaround: schedule the rewire from a different node's fn (via
    /// `Core::emit` on a state node and observing the emit downstream),
    /// or perform the rewire after the wave completes (e.g. from a sink
    /// callback that is itself outside any fn-fire scope).
    ///
    /// Slice F (2026-05-07) — A6.
    #[error(
        "set_deps({n:?}, ...): rejected — node {n:?} is currently mid-fire \
         (set_deps from inside the firing node's own fn would corrupt the \
         Dynamic `tracked` indices snapshot taken before invoke_fn). \
         Schedule the rewire outside this fire scope."
    )]
    ReentrantOnFiringNode { n: NodeId },
}

impl Core {
    /// Atomic dep mutation — change a node's upstream deps without TEARDOWN
    /// cascading and without losing cache.
    ///
    /// Per the TLA+-verified design at
    /// `~/src/graphrefly-ts/docs/research/wave_protocol_rewire.tla`
    /// (35,950 distinct states, all 7 invariants clean):
    ///
    /// - Removed deps: clear dirtyMask bit, drain pending queue, drop DepRecord.
    /// - Added deps: SENTINEL prevData; push-on-subscribe if added dep has cached DATA.
    /// - Preserved: `firstRunPassed`, `pauseLocks`, `pauseBuffer`, `cache` (ROM/RAM).
    /// - Status auto-settles if dirtyMask becomes empty.
    /// - Idempotent on `new_deps == current deps`.
    /// - Self-rewire `n ∈ new_deps` rejected (`SelfDependency`).
    /// - Cycles rejected (`WouldCreateCycle`).
    /// - Allowed mid-wave + while paused.
    /// - Phase 13.8 Q1: terminal `n` rejected (`TerminalNode`); newly-added
    ///   terminal non-resubscribable deps rejected (`TerminalDep`).
    ///
    /// The body is a single atomic dep-mutation transaction with several
    /// discrete validation stages. Splitting would require passing a
    /// partially-mutable CoreState across helpers, and the transaction's
    /// locality is what makes the F1 refcount-leak collection work.
    #[allow(clippy::too_many_lines)]
    pub fn set_deps(&self, n: NodeId, new_deps: &[NodeId]) -> Result<(), SetDepsError> {
        let mut s = self.lock_state();
        // Validate node exists and is compute. Read-once via the helper so
        // subsequent code can use `require_node(n)` without re-checking.
        let (is_state, is_producer, is_terminal) = {
            let rec = s.nodes.get(&n).ok_or(SetDepsError::UnknownNode(n))?;
            (rec.is_state(), rec.is_producer(), rec.terminal.is_some())
        };
        if is_state || is_producer {
            // State and Producer nodes have no declared deps — set_deps
            // is meaningless. Producer nodes manage their own subscriptions
            // through the binding's ProducerCtx; mutating their (empty)
            // dep set would not affect that.
            return Err(SetDepsError::NotComputeNode(n));
        }
        // Reject if `n` itself is terminal (Phase 13.8 Q1: terminal nodes
        // cannot be rewired; recovery is via resubscribable subscribe).
        if is_terminal {
            return Err(SetDepsError::TerminalNode { n });
        }
        // A6 reentrancy guard (Slice F, 2026-05-07): reject if `n` is
        // currently mid-fire on the wave-owner thread. Closes the D1 hazard
        // where `Phase 1` snapshotted `dep_handles` against pre-rewire dep
        // ordering and `Phase 3` would store the returned `tracked` indices
        // against post-rewire ordering. Same-thread re-entry is the only
        // path that matters — cross-thread emits already block on
        // `wave_owner` per the M1 design.
        if s.currently_firing.contains(&n) {
            return Err(SetDepsError::ReentrantOnFiringNode { n });
        }
        // Self-rewire rejection.
        if new_deps.contains(&n) {
            return Err(SetDepsError::SelfDependency { n });
        }
        // Validate all new deps exist.
        for &d in new_deps {
            if !s.nodes.contains_key(&d) {
                return Err(SetDepsError::UnknownNode(d));
            }
        }
        // Cycle detection: data flows parent → child via the `children` map.
        // Adding edge `d → n` (d becomes a dep of n) creates a cycle iff
        // `d` is already reachable from `n` via existing data-flow edges
        // (so `n → ... → d` exists, and the new `d → n` closes the loop).
        // DFS from `n` along `children` edges, looking for each added dep.
        let current_deps: HashSet<NodeId> = s.require_node(n).dep_ids().collect();
        let new_deps_set: HashSet<NodeId> = new_deps.iter().copied().collect();
        let added: HashSet<NodeId> = new_deps_set.difference(&current_deps).copied().collect();
        for &d in &added {
            if let Some(path) = self.path_from_to(&s, n, d) {
                return Err(SetDepsError::WouldCreateCycle {
                    n,
                    added_dep: d,
                    path,
                });
            }
        }
        // Phase 13.8 Q1: reject newly-added deps that are terminal AND not
        // resubscribable. Resubscribable terminal deps are allowed — the
        // subscribe path resets their lifecycle when something activates
        // them. Already-present (kept) deps are unaffected; their terminal
        // status was accepted at the time they terminated.
        for &d in &added {
            let dep_rec = s.require_node(d);
            if dep_rec.terminal.is_some() && !dep_rec.resubscribable {
                return Err(SetDepsError::TerminalDep { n, dep: d });
            }
        }
        // Idempotent fast-path.
        if added.is_empty() && current_deps == new_deps_set {
            return Ok(());
        }
        let removed: HashSet<NodeId> = current_deps.difference(&new_deps_set).copied().collect();

        // Snapshot old deps (ordered) for topology event, before mutation.
        let old_deps_vec: Vec<NodeId> = s.require_node(n).dep_ids_vec();

        // Carry out the rewire atomically.
        // 1. Build new dep_records, preserving DepRecord state for kept deps.
        let new_deps_vec: Vec<NodeId> = new_deps.to_vec();
        //
        // Refcount discipline (F1 audit fix): each `Some(TerminalKind::Error(h))`
        // slot owns a refcount share retained at `terminate_node` time. When a
        // dep is REMOVED, its slot is dropped — the corresponding handle's
        // share must be released here, otherwise it leaks until Core drop.
        // Also release data_batch retains for removed deps.
        let (new_dep_records, removed_handles): (Vec<DepRecord>, Vec<HandleId>) = {
            let rec = s.require_node(n);
            // Index old dep_records by NodeId for O(1) lookup of kept deps.
            let old_by_node: HashMap<NodeId, &DepRecord> =
                rec.dep_records.iter().map(|dr| (dr.node, dr)).collect();
            let new_records: Vec<DepRecord> = new_deps_vec
                .iter()
                .map(|&d| {
                    if let Some(old) = old_by_node.get(&d) {
                        // Kept dep: preserve all state (prev_data, data_batch,
                        // terminal, wave flags). Subscriptions stay live.
                        DepRecord {
                            node: d,
                            prev_data: old.prev_data,
                            dirty: old.dirty,
                            involved_this_wave: old.involved_this_wave,
                            data_batch: old.data_batch.clone(),
                            terminal: old.terminal,
                        }
                    } else {
                        // Added dep: fresh sentinel record.
                        DepRecord::new(d)
                    }
                })
                .collect();
            // Collect handles to release from REMOVED dep records.
            let mut to_release: Vec<HandleId> = Vec::new();
            for d in &removed {
                if let Some(old) = old_by_node.get(d) {
                    if let Some(TerminalKind::Error(h)) = old.terminal {
                        to_release.push(h);
                    }
                    // Release data_batch retains for removed deps.
                    for &h in &old.data_batch {
                        to_release.push(h);
                    }
                }
            }
            (new_records, to_release)
        };
        // Clear dirtyMask bit by re-emitting the wave-bookkeeping: we don't
        // currently model a per-dep dirtyMask explicitly (we use the boolean
        // `dirty` flag at node level). Removing a dep's entry from the implicit
        // mask is therefore implicit — by removing the dep, future emissions
        // from it can't re-arm the bit. The per-dep `involved_this_wave` flag
        // stays wave-scoped and gets cleared at wave end. The setDeps action
        // itself does NOT change the dirty boolean unless all deps are cleared;
        // in that case we settle.
        // Slice E2 (D067): on a dynamic node that had previously fired its
        // fn, capture `has_fired_once` BEFORE the reset so we can fire
        // `OnRerun` cleanup lock-released after `drop(s)` below. Without
        // this, the next `fire_regular` Phase 1 would capture
        // `has_fired_once = false`, causing Phase 1.5 to skip OnRerun —
        // silently dropping the prior activation's cleanup closure when
        // the next `invoke_fn` overwrites `current_cleanup`. Per spec
        // R2.4.5, `set_deps` does NOT end the activation cycle
        // (subscribe→unsubscribe is the cycle boundary), so OnRerun must
        // fire on every re-fire including post-set_deps.
        let fire_set_deps_on_rerun;
        {
            let rec = s.require_node_mut(n);
            fire_set_deps_on_rerun = rec.is_dynamic && rec.has_fired_once;
            rec.dep_records = new_dep_records;
            // Re-derive `tracked` for static derived: all indices.
            // For dynamic: clear `tracked` AND reset `has_fired_once` so the
            // next dep delivery satisfies the first-fire branch in
            // `deliver_data_to_consumer` (`!has_fired_once || tracked.contains(...)`).
            // Without resetting `has_fired_once`, the cleared `tracked` blocks
            // every future fire — fn never re-runs and the dynamic node sits
            // on stale cache derived from the old dep set. The next fire
            // re-runs fn unconditionally; fn's returned `tracked` then
            // repopulates `rec.tracked` and normal selective-deps semantics
            // resume from the next dep update onward.
            if rec.is_dynamic {
                rec.tracked.clear();
                rec.has_fired_once = false;
            } else {
                // Derived (static) and Operator track all deps.
                rec.tracked = (0..new_deps_vec.len()).collect();
            }
        }

        // 2. Update inverted-edge map (children).
        for &removed_dep in &removed {
            if let Some(set) = s.children.get_mut(&removed_dep) {
                set.remove(&n);
            }
        }
        for &added_dep in &added {
            s.children.entry(added_dep).or_default().insert(n);
        }

        // 3. Push-on-subscribe for added deps with cached DATA. Wraps in a
        // wave so any downstream propagation runs cleanly. We capture only
        // the LIST of added deps (not their cache values) because the cache
        // can change between releasing the validation lock and the wave's
        // re-acquisition — see the P2 race fix below.
        //
        // P2 (Slice A close /qa) — between `drop(s)` and `run_wave`'s
        // closure re-acquiring the lock, a concurrent thread could
        // invalidate one of the added deps, releasing its cache handle. A
        // pre-snapshot of `(added_dep, cache)` pairs would then carry a
        // dangling HandleId into `deliver_data_to_consumer`. The fix is to
        // re-read each added dep's `cache` INSIDE the closure (under the
        // freshly re-acquired state lock). The wave-owner re-entrant mutex
        // (Q2) blocks concurrent waves once we enter `run_wave`, so the
        // re-read sees a coherent post-validation state.
        let added_for_wave: Vec<NodeId> = added.iter().copied().collect();
        let added_for_registry: Vec<NodeId> = added.iter().copied().collect();
        let removed_for_registry: Vec<NodeId> = removed.iter().copied().collect();
        // Drop the state lock before run_wave (which acquires its own) and
        // before crossing the binding boundary for the F1 refcount-fix
        // releases. Keeps the lock-discipline split (binding calls outside
        // the state lock) consistent with the rest of the dispatcher.
        drop(s);
        // Slice X5 (D3 substrate, 2026-05-08): maintain partition
        // membership across topology change.
        //   - For each new edge: union the partitions of `n` and `added_dep`.
        //   - For each removed edge: notify registry (X5 commit-1: no-op;
        //     Y1 commit-2 will run reachability walk + split if disconnected).
        // Done lock-released wrt state. Registry mutex is held briefly.
        {
            let mut reg = self.registry.lock();
            for added_dep in &added_for_registry {
                reg.union_nodes(n, *added_dep);
            }
            for removed_dep in &removed_for_registry {
                reg.on_edge_removed(n, *removed_dep);
            }
        }
        // Slice E2 (D067): fire OnRerun lock-released for dynamic nodes
        // that had previously fired. The cleanup closure cleans up
        // resources tied to the old dep shape before the next fn-fire
        // (triggered by added-dep push-on-subscribe below) registers a
        // fresh cleanup spec. Direct fire (NOT via deferred_cleanup_hooks)
        // because set_deps may NOT enter a wave (no added deps → no
        // run_wave below) — queueing the hook would orphan it until the
        // next unrelated wave drains.
        if fire_set_deps_on_rerun {
            self.binding.cleanup_for(n, CleanupTrigger::OnRerun);
        }
        // Fire topology event after lock is dropped.
        self.fire_topology_event(&crate::topology::TopologyEvent::DepsChanged {
            node: n,
            old_deps: old_deps_vec,
            new_deps: new_deps_vec.clone(),
        });
        if !added_for_wave.is_empty() {
            self.run_wave(|this| {
                let mut s = this.lock_state();
                // Defensive: re-validate `n` still exists and isn't terminal.
                // A concurrent path could have terminated it between
                // validation and run_wave's wave_owner acquisition.
                if !s.nodes.contains_key(&n) || s.require_node(n).terminal.is_some() {
                    return;
                }
                for added_dep in &added_for_wave {
                    // Re-read cache under the wave-owner-held lock — this
                    // is the post-validation, post-concurrent-action
                    // snapshot. NO_HANDLE means the dep was invalidated
                    // concurrently; skip (no data to push).
                    let cache = match s.nodes.get(added_dep) {
                        Some(rec) => rec.cache,
                        None => continue, // dep deleted concurrently
                    };
                    if cache == NO_HANDLE {
                        continue;
                    }
                    let dep_idx = s.require_node(n).dep_index_of(*added_dep);
                    if let Some(idx) = dep_idx {
                        this.deliver_data_to_consumer(&mut s, n, idx, cache);
                    }
                }
            });
        }
        for h in removed_handles {
            self.binding.release_handle(h);
        }
        Ok(())
    }

    /// DFS from `from` along data-flow edges (children map) looking for `to`.
    /// Returns the path including endpoints, or `None` if unreachable. Used
    /// for cycle detection in [`Self::set_deps`].
    fn path_from_to(&self, s: &CoreState, from: NodeId, to: NodeId) -> Option<Vec<NodeId>> {
        if from == to {
            return Some(vec![from]);
        }
        let mut stack: Vec<(NodeId, Vec<NodeId>)> = vec![(from, vec![from])];
        let mut visited: HashSet<NodeId> = HashSet::new();
        while let Some((cur, path)) = stack.pop() {
            if !visited.insert(cur) {
                continue;
            }
            if cur == to {
                return Some(path);
            }
            if let Some(children) = s.children.get(&cur) {
                for &child in children {
                    let mut new_path = path.clone();
                    new_path.push(child);
                    stack.push((child, new_path));
                }
            }
        }
        None
    }
}

// CoreState helpers — kept on the inner struct so they're naturally scoped
// to the lock guard.
impl CoreState {
    fn alloc_node_id(&mut self) -> NodeId {
        let id = NodeId::new(self.next_node_id);
        self.next_node_id += 1;
        id
    }

    fn alloc_sub_id(&mut self) -> SubscriptionId {
        let id = SubscriptionId(self.next_subscription_id);
        self.next_subscription_id += 1;
        id
    }

    /// Clear wave-scoped flags and rotate per-dep batch data on every
    /// node. Run at the end of every wave (regular drain via `run_wave`,
    /// activation drain via `activate_derived`, and `BatchGuard::drop`'s
    /// drain). Centralized so a future wave-state field can't be missed
    /// at one of the cleanup sites.
    ///
    /// Per-dep rotation (R2.9.b / R1.3.6.b):
    /// - `prev_data` ← last element of `data_batch` (or unchanged if empty).
    ///   The last batch entry's retain transfers to `prev_data`; the old
    ///   `prev_data`'s retain is released. All earlier batch entries are
    ///   released.
    /// - `data_batch` cleared.
    /// - Per-dep `dirty` and `involved_this_wave` cleared.
    ///
    /// Handle releases are pushed to `deferred_handle_releases` for
    /// post-lock-drop release by the caller.
    pub(crate) fn clear_wave_state(&mut self) {
        self.pending_auto_resolve.clear();
        // A6 (Slice F, 2026-05-07): currently_firing is push/pop balanced
        // by FiringGuard's RAII (including on panic). It should already be
        // empty here, but defensively clear in case a future code path
        // forgets the guard.
        self.currently_firing.clear();
        // A3 (Slice F, 2026-05-07): pending_pause_overflow is normally
        // drained by drain_and_flush via the synthesis loop. If a wave is
        // panic-discarded BEFORE the synthesis runs (e.g. invoke_fn panics
        // before a paused-overflow has a chance to synthesize), we drop the
        // queued entries silently — the binding never sees ERROR for that
        // overflow event, but the pause buffer's `dropped` count is
        // unchanged so callers can still detect via ResumeReport. Re-firing
        // the synthesis on the next wave would be confusing (the overflow
        // event is logically scoped to the panicked wave).
        self.pending_pause_overflow.clear();
        // Slice G: tier3 emit tracking is wave-scoped.
        self.tier3_emitted_this_wave.clear();
        // Slice E2 (D057): per-wave-per-node OnInvalidate dedup is
        // wave-scoped — cleared so the next wave can fire cleanups again.
        self.invalidate_hooks_fired_this_wave.clear();
        // Slice E2 INVARIANT (DO NOT CHANGE WITHOUT THINKING):
        // `deferred_cleanup_hooks` is NOT cleared here. It follows the
        // `deferred_handle_releases` discipline:
        //   - SUCCESS path (`BatchGuard::drop` non-panic): drained by
        //     `Core::drain_deferred` AFTER `clear_wave_state` runs, then
        //     fired lock-released by `Core::fire_deferred`.
        //   - PANIC-DISCARD path (`BatchGuard::drop` panic): explicitly
        //     `std::mem::take`-and-dropped AFTER `clear_wave_state` runs,
        //     silently per D061.
        // Clearing it INSIDE `clear_wave_state` would race the success
        // path: the wave's queued `OnInvalidate` cleanup hooks would be
        // erased BEFORE `drain_deferred` could take them, dropping every
        // user cleanup callback on every successful wave.
        // If a future change moves `deferred_cleanup_hooks` ownership
        // here, ALSO move the post-`clear_wave_state` take in both
        // BatchGuard paths to BEFORE the clear call. Until then, leaving
        // the field untouched here is load-bearing.
        for rec in self.nodes.values_mut() {
            rec.dirty = false;
            rec.involved_this_wave = false;
            for dr in &mut rec.dep_records {
                let batch_len = dr.data_batch.len();
                if batch_len > 0 {
                    // Release all batch entries EXCEPT the last — the last
                    // entry's retain transfers to prev_data.
                    for &h in &dr.data_batch[..batch_len - 1] {
                        self.deferred_handle_releases.push(h);
                    }
                    // Release the OLD prev_data (its retain was from the
                    // previous wave's rotation or from initial delivery).
                    if dr.prev_data != NO_HANDLE {
                        self.deferred_handle_releases.push(dr.prev_data);
                    }
                    // Rotate: last batch entry becomes new prev_data.
                    // Its retain carries over — no extra retain needed.
                    dr.prev_data = dr.data_batch[batch_len - 1];
                    dr.data_batch.clear();
                }
                dr.dirty = false;
                dr.involved_this_wave = false;
            }
        }
    }

    pub(crate) fn require_node(&self, id: NodeId) -> &NodeRecord {
        self.nodes
            .get(&id)
            .unwrap_or_else(|| panic!("unknown node {id:?}"))
    }

    pub(crate) fn require_node_mut(&mut self, id: NodeId) -> &mut NodeRecord {
        self.nodes
            .get_mut(&id)
            .unwrap_or_else(|| panic!("unknown node {id:?}"))
    }
}

/// Release every binding-side refcount share owned by this `CoreState`
/// when the last `Core` clone drops the inner Mutex.
///
/// Without this, every retained handle in `cache` / `terminal` Error /
/// `dep_terminals` Error / pause-buffer-payload would leak in the binding
/// registry until process exit. Production bindings (napi-rs, pyo3,
/// wasm-bindgen) all maintain handle-ref maps that grow unbounded without
/// this cleanup.
///
/// Safe to call during panic unwinding — `BindingBoundary::release_handle`
/// is the only call, and a panicking binding during cleanup would already
/// have been a problem in normal operation.
impl Drop for CoreState {
    fn drop(&mut self) {
        // Drain pending in-flight retains too, so a panic mid-wave doesn't
        // strand the queue_notify retains in `deferred_handle_releases`.
        let pending = std::mem::take(&mut self.pending_notify);
        let deferred_releases = std::mem::take(&mut self.deferred_handle_releases);
        // `deferred_flush_jobs` carries `Vec<Sink>` clones — those Arcs
        // drop naturally when this CoreState drops; no handles to release
        // there.
        let _ = std::mem::take(&mut self.deferred_flush_jobs);

        // Per-node retained handles:
        //   - `cache` (1 retain per non-NO_HANDLE state cache or
        //     populated compute cache).
        //   - `terminal == Some(Error(h))` (1 retain on the terminal slot).
        //   - `dep_terminals[i] == Some(Error(h))` (1 retain per consumer's
        //     terminated-dep slot).
        //   - `pause_state` paused buffer messages with payload handles
        //     (1 retain per buffered Data/Error).
        for rec in self.nodes.values_mut() {
            if rec.cache != NO_HANDLE {
                self.binding.release_handle(rec.cache);
            }
            if let Some(TerminalKind::Error(h)) = rec.terminal {
                self.binding.release_handle(h);
            }
            for dr in &rec.dep_records {
                if let Some(TerminalKind::Error(h)) = dr.terminal {
                    self.binding.release_handle(h);
                }
                // Release data_batch retains (in-flight wave data).
                for &h in &dr.data_batch {
                    self.binding.release_handle(h);
                }
                // Release prev_data retain (cross-wave persistence).
                if dr.prev_data != NO_HANDLE {
                    self.binding.release_handle(dr.prev_data);
                }
            }
            if let PauseState::Paused { buffer, .. } = &rec.pause_state {
                for msg in buffer {
                    if let Some(h) = msg.payload_handle() {
                        self.binding.release_handle(h);
                    }
                }
            }
            // Slice E1: release replay-buffer retains.
            for &h in &rec.replay_buffer {
                self.binding.release_handle(h);
            }
            // Operator scratch (Slice C-3, D026): generic per-operator
            // state struct. Each variant's release_handles releases the
            // shares it owns (Scan/Reduce acc, Distinct/Pairwise prev,
            // Last latest + default; Take/Skip/TakeWhile own no handles).
            if let Some(scratch) = rec.op_scratch.as_mut() {
                scratch.release_handles(&*self.binding);
            }
        }

        // Pending wave retains. Slice X4 / D2: walk all batches' messages
        // — `iter_messages` flattens the per-node `SmallVec<PendingBatch>`.
        for entry in pending.values() {
            for msg in entry.iter_messages() {
                if let Some(h) = msg.payload_handle() {
                    self.binding.release_handle(h);
                }
            }
        }
        for h in deferred_releases {
            self.binding.release_handle(h);
        }
        // Wave-cache snapshot retains (defensive — should normally be
        // empty by the time Core drops, but a panicked-mid-wave Core
        // could leave them populated).
        let snapshots = std::mem::take(&mut self.wave_cache_snapshots);
        for (_, h) in snapshots {
            self.binding.release_handle(h);
        }
    }
}