polydat-core 0.6.1

// Copyright 2024-2026 Jonathan Shook
// SPDX-License-Identifier: Apache-2.0

//! The closure tier: every node runs its generated op over a flat u64
//! slot buffer, by-reference outputs as `Ref2` pairs into step-owned
//! scratch (compiled_handles.md).
//!
//! Four kernel types, each produced by a distinct compiler path. They
//! differ in what `set_inputs` marks and whether `eval_for_slot`
//! consults the cone guard; the shared step loop reads the mode's
//! `use_clean` flag per step.
//!
//! | Type | Push (per-node skip) | Pull (cone guard) |
//! |------|---------------------|-------------------|
//! | `CompiledKernelRaw` | — | — |
//! | `CompiledKernelPush` | yes | — |
//! | `CompiledKernelPull` | — | yes |
//! | `CompiledKernelPushPull` | yes | yes |

use std::collections::HashMap;

use crate::ast::{CompiledSlotOp, CompiledU64Op, PortType, ScratchBuf, ScratchElem};

/// What a P2 kernel needs beyond its steps: each named output's port
/// type for the typed reader, the externs, the provenance, and the
/// attribution for the failure path.
#[derive(Default)]
pub(crate) struct P2Extras {
    pub(crate) output_types: HashMap<String, PortType>,
    /// The kernel's extern inputs: seeded at build, host-settable,
    /// written through at every set (`compile::externs`).
    pub(crate) externs: crate::compile::externs::Externs,
    /// Per input slot, coordinates and externs alike, the steps that
    /// depend on it: the provenance the plan is derived from.
    pub(crate) input_dependents: Vec<Vec<usize>>,
    /// Where each step came from, for the failure path (engines.md §3.4).
    pub(crate) attribution: std::sync::Arc<crate::compile::Attribution>,
}

/// A single evaluation step in the compiled kernel.
/// A compiled step's op: pure-scalar u64 closure, or a slot op
/// with kernel-owned scratch for typed-slice ports
/// (type_system_alignment.md §4, compiled_handles.md §3).
pub(crate) enum StepOp {
    U64(CompiledU64Op),
    Slot(CompiledSlotOp),
    /// A slot copy (`identity`, the compiler's `__port_` passthrough),
    /// run inline: no closure call, no gather.
    Copy,
}
/// One compiled step plus its slice of the scratch arena.
pub(crate) struct P2Step {
    /// The node's name, for construction-time diagnostics.
    pub(crate) name: String,
    pub(crate) op: StepOp,
    pub(crate) input_slots: Vec<usize>,
    pub(crate) output_slots: Vec<usize>,
    /// Scratch element declarations (consumed by build_core).
    pub(crate) scratch: Vec<ScratchElem>,
    /// First slot of each Ref2-colored output port, in port
    /// order — zipped with the scratch entries to build the
    /// slot→arena map for axiom S9(a)'s validator and the S2
    /// accessors.
    pub(crate) ref_output_starts: Vec<usize>,
    /// The node handles `None` inputs itself (none_semantics.md Rule 2);
    /// every other node emits `None` when any input is `None` (Rule 1).
    pub(crate) accepts_none: bool,
    /// The node is nondeterministic or downstream of one (the runtime
    /// model's per-cycle invalidation set): never current.
    pub(crate) volatile: bool,
    /// No input reaches the node and it is not volatile: the runtime
    /// model's compile-constant lifecycle, folded at build.
    pub(crate) constant: bool,
    /// The node is a side channel: it runs exactly when the interpreter
    /// would run it, in every provenance mode, because its run is
    /// observable.
    pub(crate) side: bool,
}

struct CompiledStep {
    op: StepOp,
    input_slots: Vec<usize>,
    output_slots: Vec<usize>,
    scratch_range: (usize, usize),
    /// The closure runs on `None` inputs (none_semantics.md Rule 2).
    accepts_none: bool,
    /// Never current: nondeterministic or downstream of one.
    volatile: bool,
    /// Compile-constant: folded at build, current from then on.
    constant: bool,
    /// A side channel: skipped when current in every mode, since a
    /// redundant run would be observed.
    side: bool,
}

/// An output resolved for pulls by index: its slot, its type, its cone
/// order, and whether any step of that order can fail.
type ResolvedOutput = (
    usize,
    crate::ast::PortType,
    Option<std::sync::Arc<[usize]>>,
    bool,
);

/// Common fields shared by all kernel variants. A clone is a new state
/// of the same program: the steps are shared, everything else is the
/// clone's own (engines.md §3.5), and every pair in its buffer
/// points into its own storage (axiom S3), never into the state it was
/// cloned from.
struct KernelCore {
    /// The engine this kernel runs, as it reports it: the tier and
    /// the provenance mode it was built with. State rather than a
    /// property of the type, so one kernel type can serve a tier
    /// that runs native code and one that runs none.
    engine: crate::compile::select::Engine,
    buffer: Vec<u64>,
    coord_count: usize,
    steps: std::sync::Arc<[CompiledStep]>,
    output_map: HashMap<String, usize>,
    gather_buf: Vec<u64>,
    scatter_buf: Vec<u64>,
    /// Kernel-owned vector storage; vector-producing ports'
    /// (ptr, len) slots view entries here (type_system_alignment.md
    /// §4, compiled_handles.md §3).
    scratch: Vec<ScratchBuf>,
    /// Axiom S2: per-slot Ref2 mask — the raw readers panic on
    /// these instead of leaking addresses.
    ref_slots: Vec<bool>,
    /// Axiom S9(a): (first slot of a Ref pair → scratch arena
    /// index) for every scratch-backed Ref output.
    ref_scratch: Vec<(usize, usize)>,
    /// Port type of each named output, for `get_value`.
    output_types: HashMap<String, PortType>,
    /// The extern inputs, written through at every set.
    externs: crate::compile::externs::Externs,
    /// The traversals the program declares (for_traversal.md), opened
    /// through the `Kernel` trait.
    traversals: std::sync::Arc<[crate::dsl::traversal::Traversal]>,
    /// Per declared output, its slot, type, and cone, resolved on the
    /// first index-keyed pull (`pull_at`, runtime_model.md §6).
    resolved_outputs: Vec<Option<ResolvedOutput>>,
    /// The coordinates set through the `Kernel` trait, pending
    /// evaluation; `stale` means a write happened since the last
    /// evaluation round.
    drive: crate::compile::Drive,
    /// Per slot: the slot holds `None` (none_semantics.md Rule 1):
    /// an unset extern, or an output of a step that propagated one.
    none: Vec<bool>,
    /// Per step: the evaluation round it last ran in, so a new round
    /// forgets every run without a scan.
    ran: Vec<u64>,
    /// The evaluation round: advanced by the first evaluation after a
    /// write, so a mode without per-step currency runs a step once per
    /// round rather than once per reader. Bookkeeping only: it wipes
    /// nothing, and every output stands until an input in its
    /// provenance is written. 0 is never a round.
    epoch: u64,
    /// Every step ran in the round: a full evaluation happened.
    all_ran: bool,
    /// Per step: its outputs are current for the inputs it depends on.
    /// Cleared through the plan when an input changes, whichever call
    /// changed it; never set for a volatile step.
    clean: Vec<bool>,
    /// Whether this kernel's provenance mode skips current steps
    /// (push-side); a mode without per-step skipping runs every step
    /// in the cone once per round.
    use_clean: bool,
    /// The dirty-register plan: what each input invalidates, what each
    /// output needs.
    plan: std::sync::Arc<crate::compile::Invalidation>,
    /// Per slot: the step that writes it, for the validator.
    slot_step: std::sync::Arc<[Option<usize>]>,
    /// Where each step came from, for the failure path (engines.md §3.4).
    sites: std::sync::Arc<crate::compile::Attribution>,
    /// The step running, for the failure path.
    cur_step: usize,
    /// Every step, in order: what `eval` runs.
    all: std::sync::Arc<[usize]>,
    /// Per input slot, the steps an input change marks not current:
    /// the plan's dependents in a push mode; in a raw or pull-only
    /// mode, which never consult a pure step's currency, only the side
    /// channels among them (an optimization over the plan, not a change
    /// to it).
    dirty: std::sync::Arc<[Vec<usize>]>,
    /// The steps that are never current.
    volatile_steps: std::sync::Arc<[usize]>,
    /// Some slot holds `None`: an unset extern, which is the only way
    /// one enters (engines.md §3.3). When none does, the steps run
    /// without the mask.
    any_none: bool,
}

impl Clone for KernelCore {
    fn clone(&self) -> Self {
        let mut core = KernelCore {
            engine: self.engine,
            buffer: self.buffer.clone(),
            coord_count: self.coord_count,
            steps: self.steps.clone(),
            output_map: self.output_map.clone(),
            gather_buf: self.gather_buf.clone(),
            scatter_buf: self.scatter_buf.clone(),
            scratch: self.scratch.clone(),
            ref_slots: self.ref_slots.clone(),
            ref_scratch: self.ref_scratch.clone(),
            output_types: self.output_types.clone(),
            externs: self.externs.clone(),
            traversals: self.traversals.clone(),
            resolved_outputs: self.resolved_outputs.clone(),
            drive: self.drive.clone(),
            none: self.none.clone(),
            ran: self.ran.clone(),
            epoch: self.epoch,
            all_ran: self.all_ran,
            clean: self.clean.clone(),
            use_clean: self.use_clean,
            plan: self.plan.clone(),
            slot_step: self.slot_step.clone(),
            sites: self.sites.clone(),
            cur_step: self.cur_step,
            all: self.all.clone(),
            dirty: self.dirty.clone(),
            volatile_steps: self.volatile_steps.clone(),
            any_none: self.any_none,
        };
        core.republish_refs();
        core
    }
}

impl KernelCore {
    crate::compile::shared_core_methods!();
    /// Whether step `i` can fail: a closure runs a node's Rust body,
    /// which may panic.
    #[inline]
    fn step_can_fail(&self, _i: usize) -> bool {
        true
    }

    /// The program's identity: the step list, which every kernel created
    /// from the program and every fork shares, and no other program has.
    fn program_identity(&self) -> usize {
        std::sync::Arc::as_ptr(&self.steps) as *const () as usize
    }

    /// The program node the step now running belongs to, for the
    /// failure path (engines.md §3.4). On this tier a step is one node, so the
    /// step index is the node index. Read by `run_guarded` and by the
    /// build-time `fold_steps`, so the two always name the same node.
    #[inline]
    fn failing_node(&self) -> usize {
        self.cur_step
    }

    /// The steps of `order` that have not run in the round, in order.
    #[inline]
    fn run_order(&mut self, order: &[usize]) {
        let steps = &self.steps;
        let none_free = !self.any_none;
        for &i in order {
            if self.all_ran || self.ran[i] == self.epoch {
                continue;
            }
            let step = &steps[i];
            // A pure step may be recomputed redundantly in a mode
            // without per-step skipping; a side channel may not.
            if (self.use_clean || step.side) && self.clean[i] && !step.volatile {
                self.ran[i] = self.epoch;
                continue;
            }
            self.cur_step = i;
            if none_free {
                run_step_fast(
                    step,
                    &mut self.buffer,
                    &mut self.gather_buf,
                    &mut self.scatter_buf,
                    &mut self.scratch,
                );
            } else {
                run_step(
                    step,
                    &mut self.buffer,
                    &mut self.none,
                    &mut self.gather_buf,
                    &mut self.scatter_buf,
                    &mut self.scratch,
                );
            }
            self.ran[i] = self.epoch;
            self.clean[i] = !step.volatile;
        }
    }

    /// Every step, in order, in a round just begun, in a mode without
    /// per-step skipping and with no `None` in play: the same steps the
    /// general loop would run, without the bookkeeping a partial round
    /// needs. A current side channel is still skipped, since its run is
    /// observed.
    #[inline]
    fn run_fresh(&mut self) {
        let steps = &self.steps;
        for (i, step) in steps.iter().enumerate() {
            if step.side {
                if self.clean[i] && !step.volatile {
                    continue;
                }
                self.clean[i] = !step.volatile;
            }
            self.cur_step = i;
            run_step_fast(
                step,
                &mut self.buffer,
                &mut self.gather_buf,
                &mut self.scatter_buf,
                &mut self.scratch,
            );
        }
        self.all_ran = true;
    }

    /// Every step is a closure.
    fn plan(&self) -> crate::EnginePlan {
        crate::EnginePlan {
            closure_steps: self.steps.len(),
            ..Default::default()
        }
    }
}

/// Build kernel core from raw step data. `use_clean` is whether the
/// kernel's provenance mode skips current steps.
#[allow(clippy::too_many_arguments)]
fn build_core(
    coord_count: usize,
    total_slots: usize,
    steps: Vec<P2Step>,
    output_map: HashMap<String, usize>,
    ref_slots: Vec<bool>,
    extras: P2Extras,
    use_clean: bool,
    engine: crate::compile::select::Engine,
) -> Result<KernelCore, crate::KernelError> {
    let P2Extras {
        output_types,
        externs,
        input_dependents,
        attribution,
    } = extras;
    let max_inputs = steps.iter().map(|s| s.input_slots.len()).max().unwrap_or(0);
    let max_outputs = steps
        .iter()
        .map(|s| s.output_slots.len())
        .max()
        .unwrap_or(0);
    let mut scratch: Vec<ScratchBuf> = Vec::new();
    let mut ref_scratch: Vec<(usize, usize)> = Vec::new();
    let compiled_steps: Vec<CompiledStep> = steps
        .into_iter()
        .map(|step| {
            let start = scratch.len();
            scratch.extend(step.scratch.iter().map(|e| ScratchBuf::new(*e)));
            ref_scratch.extend(crate::compile::assembly::scratch_pairs(
                &step.name,
                &step.ref_output_starts,
                &step.scratch,
                start,
            ));
            CompiledStep {
                op: step.op,
                input_slots: step.input_slots,
                output_slots: step.output_slots,
                scratch_range: (start, scratch.len()),
                accepts_none: step.accepts_none,
                volatile: step.volatile,
                constant: step.constant,
                side: step.side,
            }
        })
        .collect();
    let mut slot_step: Vec<Option<usize>> = vec![None; total_slots];
    for (i, step) in compiled_steps.iter().enumerate() {
        for &s in &step.output_slots {
            slot_step[s] = Some(i);
        }
    }
    let step_inputs: Vec<&[usize]> = compiled_steps
        .iter()
        .map(|s| s.input_slots.as_slice())
        .collect();
    let step_outputs: Vec<&[usize]> = compiled_steps
        .iter()
        .map(|s| s.output_slots.as_slice())
        .collect();
    let plan = crate::compile::Invalidation::from_provenance(
        input_dependents,
        &step_inputs,
        &step_outputs,
        &output_map,
        total_slots,
    );
    let dirty: Vec<Vec<usize>> = plan
        .input_dependents
        .iter()
        .map(|deps| {
            if use_clean {
                deps.clone()
            } else {
                deps.iter()
                    .copied()
                    .filter(|&i| compiled_steps[i].side)
                    .collect()
            }
        })
        .collect();
    let volatile_steps: Vec<usize> = (0..compiled_steps.len())
        .filter(|&i| compiled_steps[i].volatile)
        .collect();
    let mut buffer = vec![0u64; total_slots];
    let mut none = vec![false; total_slots];
    let any_none = externs.seed(&mut buffer, Some(&mut none));
    let step_count = compiled_steps.len();
    let constants: Vec<usize> = compiled_steps
        .iter()
        .enumerate()
        .filter(|(_, s)| s.constant)
        .map(|(i, _)| i)
        .collect();
    let mut core = KernelCore {
        engine,
        buffer,
        coord_count,
        steps: compiled_steps.into(),
        output_map,
        gather_buf: vec![0u64; max_inputs],
        scatter_buf: vec![0u64; max_outputs],
        scratch,
        ref_slots,
        ref_scratch,
        output_types,
        externs,
        traversals: Vec::new().into(),
        resolved_outputs: Vec::new(),
        drive: crate::compile::Drive {
            coords: Vec::new(),
            stale: true,
        },
        none,
        ran: vec![0; step_count],
        epoch: 0,
        all_ran: false,
        clean: vec![false; step_count],
        use_clean,
        plan: std::sync::Arc::new(plan),
        slot_step: slot_step.into(),
        sites: attribution,
        cur_step: 0,
        all: (0..step_count).collect::<Vec<usize>>().into(),
        dirty: dirty.into(),
        volatile_steps: volatile_steps.into(),
        any_none,
    };
    // The compile-constant fold of the runtime model, on this engine: a
    // step no input reaches runs at build, once, and is current from
    // then on, so what is knowable at build is known at build and fails
    // at build.
    core.begin_epoch();
    core.fold_steps(&constants)?;
    core.drive.stale = true;
    Ok(core)
}

/// The provenance of every slot, from the steps' output slots
/// ([`crate::compile::slot_provenance`]).
fn compute_slot_provenance(
    coord_count: usize,
    total_slots: usize,
    input_dependents: &[Vec<usize>],
    steps: &[CompiledStep],
) -> Vec<crate::kernel::ProvMask> {
    let outs: Vec<&[usize]> = steps.iter().map(|s| s.output_slots.as_slice()).collect();
    crate::compile::slot_provenance(coord_count, total_slots, &outs, input_dependents)
}

// ── The writes, which the two tiers spell differently ──────────

/// Setting an extern and dirtying what it reaches. The hybrid writes
/// these itself: its `mark_input_changed` walks a plan it rebuilds when
/// `use_clean` changes, and its `mark_all_dirty` is empty on the mode
/// that runs every step anyway. Everything a compiled kernel reads
/// rather than writes is [`crate::compile::kernel_accessors`], shared
/// with the hybrid.
macro_rules! closure_writes {
    () => {
        /// Set an extern by name, as `PolydatState::set_input` does on
        /// the interpreter. The value must be of the declared port
        /// type. Every kind is written through at once, and every step
        /// downstream of the extern reruns.
        pub fn set_input(
            &mut self,
            name: &str,
            value: crate::ast::Value,
        ) -> Result<(), crate::kernel::WriteError> {
            let slot = self.core.set_extern(name, value)?;
            self.mark_input_changed(slot);
            Ok(())
        }

        /// [`Self::set_input`] by input index.
        pub fn set_input_at(
            &mut self,
            index: usize,
            value: crate::ast::Value,
        ) -> Result<(), crate::kernel::WriteError> {
            let slot = self.core.set_extern_at(index, value)?;
            self.mark_input_changed(slot);
            Ok(())
        }

        /// Every step downstream of a coordinate reruns at the next
        /// evaluation: the state a kernel created from a shared program
        /// starts in.
        fn mark_all_dirty(&mut self) {
            for i in 0..self.core.coord_count {
                self.mark_input_changed(i);
            }
        }
    };
}

// ═══════════════════════════════════════════════════════════════
// Raw: no provenance, no cone guard. Eval runs all steps.
// ═══════════════════════════════════════════════════════════════

#[derive(Clone)]
/// The closure tier with no provenance: every evaluation runs every step.
pub struct CompiledKernelRaw {
    core: KernelCore,
}

impl CompiledKernelRaw {
    pub(crate) fn new(
        coord_count: usize,
        total_slots: usize,
        steps: Vec<P2Step>,
        output_map: HashMap<String, usize>,
        ref_slots: Vec<bool>,
        extras: P2Extras,
    ) -> Result<Self, crate::KernelError> {
        Ok(Self {
            core: build_core(
                coord_count,
                total_slots,
                steps,
                output_map,
                ref_slots,
                extras,
                false,
                Engine::Closures(Provenance::Raw),
            )?,
        })
    }

    /// The plan invalidates what depends on the input; this mode runs
    /// every step of a cone once per round regardless.
    fn mark_input_changed(&mut self, slot: usize) {
        self.core.dirty_input(slot);
    }

    /// The coordinates, written; a changed one invalidates
    /// its dependents through the plan, as in every mode.
    #[inline]
    fn set_coords(&mut self, coords: &[u64]) {
        for (i, &c) in coords
            .iter()
            .enumerate()
            .take(self.core.externs.coordinate_slots())
        {
            if self.core.buffer[i] != c {
                self.core.buffer[i] = c;
                self.core.dirty_input(i);
            }
        }
    }

    /// Evaluate every step for `coords`: a new round.
    #[inline]
    pub fn eval(&mut self, coords: &[u64]) {
        self.set_coords(coords);
        self.core.drive.stale = true;
        self.core.eval_all();
    }

    /// Eval + return a specific slot. No cone guard — always evaluates.
    #[inline]
    pub fn eval_for_slot(&mut self, coords: &[u64], slot: usize) -> u64 {
        self.core.guard_ref_slot(slot);
        self.eval(coords);
        self.core.buffer[slot]
    }

    crate::compile::kernel_accessors!(set_coords);
    closure_writes!();
}

// ═══════════════════════════════════════════════════════════════
// Push: per-step skip, no cone guard. A changed input invalidates
// its dependents through the plan; a current step is skipped.
// ═══════════════════════════════════════════════════════════════

#[derive(Clone)]
/// The closure tier with per-step skipping: a changed input invalidates
/// its dependents through the plan, and a current step is skipped.
pub struct CompiledKernelPush {
    core: KernelCore,
}

impl CompiledKernelPush {
    pub(crate) fn new(
        coord_count: usize,
        total_slots: usize,
        steps: Vec<P2Step>,
        output_map: HashMap<String, usize>,
        input_dependents: Vec<Vec<usize>>,
        ref_slots: Vec<bool>,
        extras: P2Extras,
    ) -> Result<Self, crate::KernelError> {
        // The plan in `extras` carries the dependents.
        let _ = input_dependents;
        Ok(Self {
            core: build_core(
                coord_count,
                total_slots,
                steps,
                output_map,
                ref_slots,
                extras,
                true,
                Engine::Closures(Provenance::Push),
            )?,
        })
    }

    #[inline]
    fn set_coords(&mut self, coords: &[u64]) {
        for (i, &c) in coords
            .iter()
            .enumerate()
            .take(self.core.externs.coordinate_slots())
        {
            if self.core.buffer[i] != c {
                self.core.buffer[i] = c;
                self.core.dirty_input(i);
            }
        }
    }

    /// Every step downstream of the slot reruns.
    fn mark_input_changed(&mut self, slot: usize) {
        self.core.dirty_input(slot);
    }

    /// Evaluate every step that is not current for `coords`: a new round.
    #[inline]
    pub fn eval(&mut self, coords: &[u64]) {
        self.set_coords(coords);
        self.core.drive.stale = true;
        self.core.eval_all();
    }

    /// Eval + return a specific slot. No cone guard — always enters eval loop.
    #[inline]
    pub fn eval_for_slot(&mut self, coords: &[u64], slot: usize) -> u64 {
        self.core.guard_ref_slot(slot);
        self.eval(coords);
        self.core.buffer[slot]
    }

    crate::compile::kernel_accessors!(set_coords);
    closure_writes!();
}

// ═══════════════════════════════════════════════════════════════
// Pull: cone guard only, no per-step skip.
// set_inputs tracks changed_mask. eval_for_slot checks cone
// then runs ALL steps if dirty.
// ═══════════════════════════════════════════════════════════════

#[derive(Clone)]
/// The closure tier with the cone guard: an output whose cone no changed
/// input reaches is not recomputed.
pub struct CompiledKernelPull {
    core: KernelCore,
    slot_provenance: Vec<crate::kernel::ProvMask>,
    changed_mask: crate::kernel::ProvMask,
    /// Set by `set_input`: an extern changed, so the next evaluation
    /// runs whatever the cone guard says.
    force_run: bool,
}

impl CompiledKernelPull {
    pub(crate) fn new(
        coord_count: usize,
        total_slots: usize,
        steps: Vec<P2Step>,
        output_map: HashMap<String, usize>,
        input_dependents: &[Vec<usize>],
        ref_slots: Vec<bool>,
        extras: P2Extras,
    ) -> Result<Self, crate::KernelError> {
        let core = build_core(
            coord_count,
            total_slots,
            steps,
            output_map,
            ref_slots,
            extras,
            false,
            Engine::Closures(Provenance::Pull),
        )?;
        let slot_provenance =
            compute_slot_provenance(coord_count, total_slots, input_dependents, &core.steps);
        Ok(Self {
            core,
            slot_provenance,
            changed_mask: crate::kernel::ProvMask::all_below(coord_count), // all dirty initially
            force_run: false,
        })
    }

    /// Track which inputs changed (for the cone guard), and invalidate
    /// their dependents through the plan, as in every mode.
    #[inline]
    fn set_coords(&mut self, coords: &[u64]) {
        self.changed_mask.clear();
        for (i, &c) in coords
            .iter()
            .enumerate()
            .take(self.core.externs.coordinate_slots())
        {
            if self.core.buffer[i] != c {
                self.core.buffer[i] = c;
                self.changed_mask.set(i);
                self.core.dirty_input(i);
            }
        }
    }

    /// The next evaluation runs regardless of the cone guard, since
    /// `set_inputs` rebuilds the changed set from the coordinates alone.
    fn mark_input_changed(&mut self, slot: usize) {
        self.core.dirty_input(slot);
        self.force_run = true;
    }

    /// Evaluate eagerly (no cone guard). Runs all steps: a new round.
    #[inline]
    pub fn eval(&mut self, coords: &[u64]) {
        self.set_coords(coords);
        self.force_run = false;
        self.core.drive.stale = true;
        self.core.eval_all();
    }

    /// Cone guard: if the output's cone is clean, skip eval entirely.
    /// Otherwise run ALL steps (no per-node skip).
    #[inline]
    pub fn eval_for_slot(&mut self, coords: &[u64], slot: usize) -> u64 {
        self.core.guard_ref_slot(slot);
        self.set_coords(coords);
        if !self.force_run
            && slot < self.slot_provenance.len()
            && !self.slot_provenance[slot].intersects(&self.changed_mask)
        {
            return self.core.buffer[slot];
        }
        self.force_run = false;
        self.core.drive.stale = true;
        self.core.eval_all();
        self.core.buffer[slot]
    }

    crate::compile::kernel_accessors!(set_coords);
    closure_writes!();
}

// ═══════════════════════════════════════════════════════════════
// PushPull: push-side per-step skip + pull-side cone guard.
// Full optimization.
// ═══════════════════════════════════════════════════════════════

#[derive(Clone)]
/// The closure tier with per-step skipping and the cone guard.
pub struct CompiledKernelPushPull {
    core: KernelCore,
    slot_provenance: Vec<crate::kernel::ProvMask>,
    changed_mask: crate::kernel::ProvMask,
    /// Set by `set_input`: an extern changed, so the next evaluation
    /// runs whatever the cone guard says.
    force_run: bool,
}

impl CompiledKernelPushPull {
    pub(crate) fn new(
        coord_count: usize,
        total_slots: usize,
        steps: Vec<P2Step>,
        output_map: HashMap<String, usize>,
        input_dependents: Vec<Vec<usize>>,
        ref_slots: Vec<bool>,
        extras: P2Extras,
    ) -> Result<Self, crate::KernelError> {
        let core = build_core(
            coord_count,
            total_slots,
            steps,
            output_map,
            ref_slots,
            extras,
            true,
            Engine::Closures(Provenance::PushPull),
        )?;
        let slot_provenance =
            compute_slot_provenance(coord_count, total_slots, &input_dependents, &core.steps);
        Ok(Self {
            core,
            slot_provenance,
            changed_mask: crate::kernel::ProvMask::all_below(coord_count),
            force_run: false,
        })
    }

    #[inline]
    fn set_coords(&mut self, coords: &[u64]) {
        self.changed_mask.clear();
        for (i, &c) in coords
            .iter()
            .enumerate()
            .take(self.core.externs.coordinate_slots())
        {
            if self.core.buffer[i] != c {
                self.core.buffer[i] = c;
                self.changed_mask.set(i);
                self.core.dirty_input(i);
            }
        }
    }

    /// Every step downstream of the slot reruns, and the next
    /// evaluation runs whatever the cone guard says.
    fn mark_input_changed(&mut self, slot: usize) {
        self.core.dirty_input(slot);
        self.force_run = true;
    }

    /// Eval with push-side skip (no cone guard): a new round.
    #[inline]
    pub fn eval(&mut self, coords: &[u64]) {
        self.set_coords(coords);
        self.force_run = false;
        self.core.drive.stale = true;
        self.core.eval_all();
    }

    /// Cone guard + push-side skip: the full optimization.
    #[inline]
    pub fn eval_for_slot(&mut self, coords: &[u64], slot: usize) -> u64 {
        self.core.guard_ref_slot(slot);
        self.set_coords(coords);
        if !self.force_run
            && slot < self.slot_provenance.len()
            && !self.slot_provenance[slot].intersects(&self.changed_mask)
        {
            return self.core.buffer[slot];
        }
        self.force_run = false;
        self.core.drive.stale = true;
        self.core.eval_all();
        self.core.buffer[slot]
    }

    crate::compile::kernel_accessors!(set_coords);
    closure_writes!();
}

// ── The engine-independent surface (engines.md §3.5) ──────

use crate::compile::select::{Engine, Provenance};

crate::compile::impl_kernel_trait!(CompiledKernelRaw);
crate::compile::impl_kernel_trait!(CompiledKernelPush);
crate::compile::impl_kernel_trait!(CompiledKernelPull);
crate::compile::impl_kernel_trait!(CompiledKernelPushPull);
crate::compile::impl_slot_kernel!(CompiledKernelRaw);
crate::compile::impl_slot_kernel!(CompiledKernelPush);
crate::compile::impl_slot_kernel!(CompiledKernelPull);
crate::compile::impl_slot_kernel!(CompiledKernelPushPull);

/// One step: none_semantics.md Rule 1, then gather, run the closure,
/// scatter. A
/// node that does not accept `None` emits `None` on every output when
/// any input is `None`, without running.
#[inline(always)]
fn run_step(
    step: &CompiledStep,
    buffer: &mut [u64],
    none: &mut [bool],
    gather: &mut [u64],
    scatter: &mut [u64],
    scratch: &mut [ScratchBuf],
) {
    let mut any_none = false;
    for (i, &s) in step.input_slots.iter().enumerate() {
        gather[i] = buffer[s];
        any_none |= none[s];
    }
    if any_none && !step.accepts_none {
        for &s in &step.output_slots {
            none[s] = true;
        }
        return;
    }
    if matches!(step.op, StepOp::Copy) {
        for (&i, &o) in step.input_slots.iter().zip(&step.output_slots) {
            buffer[o] = buffer[i];
            none[o] = false;
        }
        return;
    }
    let (n_in, n_out) = (step.input_slots.len(), step.output_slots.len());
    match &step.op {
        StepOp::Copy => unreachable!(),
        StepOp::U64(op) => op(&gather[..n_in], &mut scatter[..n_out]),
        StepOp::Slot(op) => op(
            &gather[..n_in],
            &mut scatter[..n_out],
            &mut scratch[step.scratch_range.0..step.scratch_range.1],
        ),
    }
    for (i, &s) in step.output_slots.iter().enumerate() {
        buffer[s] = scatter[i];
        none[s] = false;
    }
}

/// [`run_step`] when no slot holds `None`: gather, run, scatter.
#[inline(always)]
fn run_step_fast(
    step: &CompiledStep,
    buffer: &mut [u64],
    gather: &mut [u64],
    scatter: &mut [u64],
    scratch: &mut [ScratchBuf],
) {
    if matches!(step.op, StepOp::Copy) {
        for (&i, &o) in step.input_slots.iter().zip(&step.output_slots) {
            buffer[o] = buffer[i];
        }
        return;
    }
    for (i, &s) in step.input_slots.iter().enumerate() {
        gather[i] = buffer[s];
    }
    let (n_in, n_out) = (step.input_slots.len(), step.output_slots.len());
    match &step.op {
        StepOp::Copy => unreachable!(),
        StepOp::U64(op) => op(&gather[..n_in], &mut scatter[..n_out]),
        StepOp::Slot(op) => op(
            &gather[..n_in],
            &mut scatter[..n_out],
            &mut scratch[step.scratch_range.0..step.scratch_range.1],
        ),
    }
    for (i, &s) in step.output_slots.iter().enumerate() {
        buffer[s] = scatter[i];
    }
}