polydat-core 0.6.1

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

//! Extern on the compiled engines.
//!
//! An `extern` is a typed input slot with a default that a host may
//! overwrite. The interpreter keeps it as a `Value` in the state; the
//! compiled engines keep every input in the flat slot buffer, so an
//! extern needs a slot, an encoding of its value into that slot, and a
//! way for the host to change it. This module is that plumbing, shared
//! by the closure, hybrid, and native kernels:
//!
//! - **Layout.** Every input beyond the coordinates owns its slots in
//!   the buffer, after the coordinates, at the width its port type
//!   names. A `Ref2`-kind extern (a string, a byte string, JSON, an
//!   extension value, a handle) owns two slots: the pair into the
//!   value it stores.
//! - **Writing through.** Every write of an extern reaches the buffer
//!   at once: a carrier (u64, i64, f64, bool) as its bits, a `Ref2`
//!   kind (a string, a byte string, JSON, an extension value, a
//!   handle) as a `(ptr, len)` pair into the value the extern stores,
//!   which stands until the host writes the extern again (axioms S3,
//!   S4). Nothing is rewritten on any other occasion: an extern is an
//!   input, and its slot stands until it is set.
//! - **Setting.** `set` type-checks the value against the declared
//!   port type, stores it, and writes it through; the kernel that owns
//!   the buffer marks the slot's dependents dirty.
//! - **Cells.** A `shared` binding's slot is bound to a `SharedCell`
//!   (engines.md §3.6), the same cell type the interpreter
//!   attaches: the cell is the register. `set` publishes through it,
//!   every run and every pull refresh the slot from it when its
//!   revision moved, and a host attaches one kernel's cell to another
//!   through the `Kernel` trait so both read and write one register.

use std::collections::HashMap;

use crate::ast::SlotShape;
use crate::ast::{PortType, Value};
use crate::kernel::InputDef;
use crate::kernel::WriteError;

/// One extern input of a compiled kernel.
#[derive(Clone)]
pub(crate) struct ExternSlot {
    pub name: String,
    /// First buffer slot: one slot for a carrier, two for a `Ref2`
    /// kind's pair.
    pub slot: usize,
    pub ty: PortType,
    /// The type the slot reports: `ty`, or for a converted input the type
    /// its readers see, where `ty` is `Dyn` (input_variance.md §5).
    pub reported: PortType,
    /// The current value: the declared default until the host sets it.
    /// A `Ref2` kind's pair points into this value, so it is written
    /// only through `set_slot`, which republishes the pair.
    pub value: Value,
    /// The declared default, what a kernel created from the program
    /// starts with.
    pub default: Value,
    /// The shared cell a `shared` binding's slot is bound to.
    pub cell: Option<crate::kernel::SharedCell>,
    /// The cell revision the slot last took its value from.
    pub seen: Option<u64>,
    /// The slot holds a `const` binding's value, which only
    /// initialization writes.
    pub is_const: bool,
    /// Only initialization reads the slot, from Rust: a const's slot, a
    /// const's fallback input, or a `shared` register with a computed
    /// start that nothing has seeded yet. Native code reads the rest,
    /// so only the rest count toward `Externs::unset_read`.
    pub init_only: bool,
    /// The slot is a `shared` register with a computed start. It is
    /// `init_only` until a write or a cell refresh seeds it, and from
    /// then on counts as any extern native code reads.
    pub register_start: bool,
}

impl ExternSlot {
    /// Whether the slot counts toward `Externs::unset_read`.
    #[inline]
    fn counted_unset(&self) -> bool {
        !self.init_only && self.value == Value::None
    }

    /// The register has a value from a write or a published cell:
    /// native code reads it from now on.
    #[inline]
    fn seeded(&mut self) {
        if self.register_start {
            self.init_only = false;
        }
    }
}

/// Whether only initialization reads the extern `name`: a const's slot
/// or a const's fallback input.
fn init_only(name: &str, is_const: bool, inits: &[crate::kernel::ConstInit]) -> bool {
    is_const
        || inits
            .iter()
            .any(|c| c.slot == name || c.fallback.as_deref() == Some(name))
}

/// Whether the extern `name` is a `shared` register with a computed
/// start.
fn register_start(name: &str, inits: &[crate::kernel::ConstInit]) -> bool {
    inits.iter().any(|c| c.register && c.slot == name)
}

/// Per input of a program, whether it is an extern native code reads
/// that a host can leave with no value: the rule of
/// [`Externs::unset_read_slots`], by input index, for an engine that
/// fuses before it has an extern set.
#[cfg(feature = "jit")]
pub(crate) fn unset_read_inputs(
    input_defs: &[InputDef],
    coord_count: usize,
    inits: &[crate::kernel::ConstInit],
) -> Vec<bool> {
    input_defs
        .iter()
        .enumerate()
        .map(|(i, def)| {
            let is_const = def.kind == crate::kernel::InputKind::Const;
            i >= coord_count
                && (!init_only(&def.name, is_const, inits) || register_start(&def.name, inits))
        })
        .collect()
}

/// The parts of an extern set that only a *composed* program uses:
/// what a binder asks of a parent, and the cells a descendant reads
/// through. A program nobody built a subscope under carries all three
/// empty.
///
/// Behind one pointer because `Externs` is embedded **by value** in all
/// three compiled cores (`KernelCore`, `HybridCore`, `JitCore`): inline,
/// the three cost about a hundred bytes of every core, and `Externs`
/// measures 328 bytes unboxed against 224 boxed.
///
/// The indirection keeps cold data apart from hot data; it is not a
/// speedup on its own. On `engine_ladder`, the boxed and unboxed forms
/// differ by less than the `p1_interpreter` canary's noise on every
/// native rung. Growing `Externs` by value does cost: about forty bytes
/// more, with a rare cell-refresh path inline beside it, measured about
/// 5% slower on the `p3_native_dyn` rung in paired rounds. That is why
/// the per-input origins live here, the coordinate counts are `u32`,
/// and the refresh work is out of line. Put what is not read per cycle
/// in here, not beside it.
#[derive(Default)]
struct ScopeCells {
    /// The binding modifiers of the named outputs, so a compiled
    /// kernel can answer what a binder asks of a parent without
    /// keeping a `PolydatProgram` to ask.
    output_modifiers: HashMap<String, crate::dsl::ast::BindingModifier>,
    /// Cells carried forward for a descendant, held by no slot of this
    /// kernel's own: the interpreter's transit, on the compiled side.
    transit_cells: Vec<crate::kernel::SharedCellEntry>,
    /// Keyed by output slot, the broadcast cell a descendant asked for,
    /// made on the first ask and not before (cross_fiber_invalidation.md
    /// §3.1, "compiled kernels, broadcast outputs").
    output_cells: std::sync::Mutex<Vec<Option<crate::kernel::SharedCell>>>,
    /// The Rule 2 write-throughs this kernel commits, as
    /// `(export_name, source_output)`: what a scope module hands the
    /// kernels it instantiates (native_scope_trees.md §3).
    write_throughs: Vec<(String, String)>,
    /// Per input, how its type was established (input_variance.md §3):
    /// read by `input_type_origin`, never per cycle.
    origins: Vec<crate::kernel::TypeOrigin>,
    /// The `const` bindings a kernel initializes, in dependency order.
    const_inits: Vec<crate::kernel::ConstInit>,
    /// The outputs whose value is fixed for a kernel's life: consts and
    /// values folded at build. `folded_value` answers only for these.
    fixed_outputs: std::collections::HashSet<String>,
    /// The outputs the program re-exports for its descendants without
    /// owning them, sorted: marked by the subscope builder, part of the
    /// canonical hash (scope_model.md §8).
    inherited_outputs: Vec<String>,
}

/// The extern inputs of one compiled kernel.
#[derive(Default)]
pub(crate) struct Externs {
    slots: Vec<ExternSlot>,
    by_name: HashMap<String, usize>,
    /// Every input by name, the coordinates first, as the interpreter
    /// program lists them.
    input_names: Vec<String>,
    /// How many inputs are coordinates (they come first).
    coordinates: u32,
    /// How many buffer slots the coordinates occupy, from slot 0: what
    /// `set_inputs` writes, a word per slot.
    coordinate_slots: u32,
    /// How many externs that native code reads have no value: kept
    /// current by every write and cell refresh, so a pure native run
    /// checks this one count rather than scanning the externs.
    unset_read: u32,
    /// Per input index, the extern slot it names; `None` for a
    /// coordinate. The index-keyed set (`set_input_at`,
    /// runtime_model.md §6).
    by_index: Vec<Option<usize>>,
    /// Every named output in declaration order, as the interpreter
    /// program lists them.
    output_names: Vec<String>,
    /// The cursors the program declares (engines.md §3.5):
    /// each is an `Ext` extern plus six scalar ones, and its schema
    /// carries the partitions the compiler resolved at build.
    cursors: Vec<crate::iteration::source::SourceSchema>,
    /// What only a composed program uses, behind one pointer
    /// ([`ScopeCells`]).
    scope: Box<ScopeCells>,
    /// This kernel's intent-dirty word, shared by the cells it seeds
    /// (cross_fiber_invalidation.md §3.1).
    intent: std::sync::Arc<std::sync::atomic::AtomicU64>,
    /// The next bit of `intent` to give a cell.
    next_bit: std::sync::atomic::AtomicU8,
    /// Whether a descendant has asked for a broadcast cell: the one
    /// lock-free check every pull makes before publishing. A clone is a
    /// new state with no descendant, so it starts false.
    broadcasting: std::sync::atomic::AtomicBool,
    /// Slots whose value a cell refresh changed, for the kernel to mark
    /// dirty; drained after every refresh.
    changed: Vec<usize>,
    /// The compile ledger of the tree this kernel's program belongs
    /// to, recorded in at build: what the compiled engines keep of
    /// the program tree's identity.
    ledger: std::sync::Arc<crate::kernel::CompileLedger>,
    /// The resource scope of the tree this kernel's program belongs to,
    /// shared by every clone.
    resources: crate::resource::ResourceScope,
    /// The digest of the graph the compiler resolved for this kernel's
    /// program, before the engine lowered it: the graph part of the
    /// program's canonical hash.
    graph_identity: [u8; 32],
}

/// Everything a clone carries but the broadcast cells, which it does
/// not: a clone is a new state of the same program (engines.md §3.5),
/// and a new state has no descendant bound to it. Sharing them would
/// have two states publishing into one register, so a descendant of
/// the original would read whichever pulled last.
///
/// The `shared` slots' cells are a different matter and are shared, as
/// they were before: a `shared` binding *is* one register, and
/// `reset_to_program` is what gives a kernel cells of its own.
impl Clone for Externs {
    fn clone(&self) -> Self {
        Self {
            slots: self.slots.clone(),
            by_name: self.by_name.clone(),
            input_names: self.input_names.clone(),
            coordinates: self.coordinates,
            coordinate_slots: self.coordinate_slots,
            unset_read: self.unset_read,
            by_index: self.by_index.clone(),
            output_names: self.output_names.clone(),
            cursors: self.cursors.clone(),
            scope: Box::new(ScopeCells {
                output_modifiers: self.scope.output_modifiers.clone(),
                transit_cells: self.scope.transit_cells.clone(),
                output_cells: std::sync::Mutex::new(Vec::new()),
                write_throughs: self.scope.write_throughs.clone(),
                origins: self.scope.origins.clone(),
                const_inits: self.scope.const_inits.clone(),
                fixed_outputs: self.scope.fixed_outputs.clone(),
                inherited_outputs: self.scope.inherited_outputs.clone(),
            }),
            intent: self.intent.clone(),
            next_bit: std::sync::atomic::AtomicU8::new(
                self.next_bit.load(std::sync::atomic::Ordering::Relaxed),
            ),
            broadcasting: std::sync::atomic::AtomicBool::new(false),
            changed: self.changed.clone(),
            ledger: self.ledger.clone(),
            resources: self.resources.clone(),
            graph_identity: self.graph_identity,
        }
    }
}

impl Externs {
    /// The externs among `input_defs` (every input after the first
    /// `coord_count`), each at the slot `input_starts` gives it. An
    /// extern is a one-slot carrier or a `Ref2` kind; a two-slot
    /// immediate has no compiled form.
    pub(crate) fn new(
        input_defs: &[InputDef],
        coord_count: usize,
        input_starts: &[usize],
        cursors: &[crate::iteration::source::SourceSchema],
        shared: &[&str],
        ledger: std::sync::Arc<crate::kernel::CompileLedger>,
    ) -> Result<Self, String> {
        let mut slots = Vec::new();
        let mut by_name = HashMap::new();
        let mut by_index = vec![None; input_defs.len()];
        for (i, def) in input_defs.iter().enumerate().skip(coord_count) {
            if def.port_type.slot_color() == crate::ast::SlotColor::Imm2 {
                return Err(format!(
                    "extern '{}' has type {}, a two-slot immediate; the compiled engines carry \
                     one-slot carriers and by-reference externs (strings, byte strings, JSON, \
                     extension values, handles)",
                    def.name, def.port_type,
                ));
            }
            by_name.insert(def.name.clone(), slots.len());
            by_index[i] = Some(slots.len());
            slots.push(ExternSlot {
                name: def.name.clone(),
                slot: input_starts[i],
                ty: def.port_type,
                reported: def.converts_to.unwrap_or(def.port_type),
                value: def.default.clone(),
                default: def.default.clone(),
                cell: None,
                seen: None,
                is_const: def.kind == crate::kernel::InputKind::Const,
                init_only: def.kind == crate::kernel::InputKind::Const,
                register_start: false,
            });
        }
        let mut externs = Self {
            slots,
            by_name,
            scope: Box::default(),
            input_names: input_defs.iter().map(|d| d.name.clone()).collect(),
            coordinates: coord_count as u32,
            coordinate_slots: input_defs
                .iter()
                .take(coord_count)
                .map(|d| crate::ast::SlotShape::slot_width(&d.port_type))
                .sum::<usize>() as u32,
            unset_read: 0,
            by_index,
            output_names: Vec::new(),
            cursors: cursors.to_vec(),
            intent: std::sync::Arc::new(std::sync::atomic::AtomicU64::new(0)),
            next_bit: std::sync::atomic::AtomicU8::new(0),
            broadcasting: std::sync::atomic::AtomicBool::new(false),
            changed: Vec::new(),
            ledger,
            resources: crate::resource::ResourceScope::new(),
            graph_identity: [0; 32],
        };
        externs.scope.origins = input_defs.iter().map(|d| d.type_origin).collect();
        for name in shared {
            if let Some(&i) = externs.by_name.get(*name) {
                let cell = externs.new_cell(externs.slots[i].value.clone());
                externs.slots[i].seen = Some(cell.snapshot().1);
                externs.slots[i].cell = Some(cell);
            }
        }
        externs.recount_unset();
        // One compiled program of the tree, whichever engine it is on.
        externs.ledger.record();
        Ok(externs)
    }

    /// No externs, and `coordinates` coordinates: a kernel assembled
    /// from steps directly rather than from a program's inputs.
    #[cfg(feature = "jit")]
    pub(crate) fn coordinates_only(coordinates: usize) -> Self {
        Self {
            coordinates: coordinates as u32,
            coordinate_slots: coordinates as u32,
            ..Self::default()
        }
    }

    /// How many buffer slots the coordinates occupy, from slot 0: the
    /// words `set_inputs` writes. Not the core's `coord_count`, which
    /// spans every input's slots, externs included.
    #[inline]
    pub(crate) fn coordinate_slots(&self) -> usize {
        self.coordinate_slots as usize
    }

    /// The compile ledger of the tree this kernel's program belongs to.
    pub(crate) fn ledger(&self) -> &std::sync::Arc<crate::kernel::CompileLedger> {
        &self.ledger
    }

    /// The resource scope of the tree this kernel's program belongs to.
    pub(crate) fn resources(&self) -> &crate::resource::ResourceScope {
        &self.resources
    }

    /// Record the resource scope of the tree this kernel's program
    /// belongs to, once, at build.
    pub(crate) fn set_resources(&mut self, resources: crate::resource::ResourceScope) {
        self.resources = resources;
    }

    /// The graph digest of this kernel's program.
    pub(crate) fn graph_identity(&self) -> [u8; 32] {
        self.graph_identity
    }

    /// Record the graph digest of this kernel's program, once, at build.
    pub(crate) fn set_graph_identity(&mut self, digest: [u8; 32]) {
        self.graph_identity = digest;
    }

    /// The outputs this kernel's program re-exports without owning
    /// them, sorted.
    pub(crate) fn inherited_outputs(&self) -> &[String] {
        &self.scope.inherited_outputs
    }

    /// Mark `names` as outputs the program re-exports without owning
    /// them, before the program is shared.
    pub(crate) fn set_inherited_outputs(&mut self, mut names: Vec<String>) {
        names.sort();
        names.dedup();
        self.scope.inherited_outputs = names;
    }

    /// The broadcast cell for the output at `slot`, made on the first
    /// ask and holding `initial` then. A later ask returns the same
    /// cell, so every descendant that binds to this output binds to one
    /// register.
    /// Behind a mutex, and taking `&self`, because asking a parent for
    /// a cell is a read from the caller's side: a binder holds the
    /// parent by shared reference while it wires a child, and the
    /// parent's evaluation state is not what changes. The interpreter
    /// answers the same question from cells it seeded at construction,
    /// through `&self`, and the two surfaces should not differ in that.
    pub(crate) fn output_cell(&self, slot: usize, initial: Value) -> crate::kernel::SharedCell {
        let bit = {
            let cells = self
                .scope
                .output_cells
                .lock()
                .expect("output cells poisoned");
            if let Some(Some(cell)) = cells.get(slot) {
                return cell.clone();
            }
            // The bit is drawn before the lock on the cell list is
            // dropped so two threads asking at once cannot take the
            // same one.
            self.next_cell_bit()
        };
        let cell = std::sync::Arc::new(crate::kernel::SharedCellInner::new(
            initial,
            self.intent.clone(),
            bit,
        ));
        let mut cells = self
            .scope
            .output_cells
            .lock()
            .expect("output cells poisoned");
        if cells.len() <= slot {
            cells.resize(slot + 1, None);
        }
        let cell = cells[slot].get_or_insert(cell).clone();
        // Set while the lock is held, so a pull that sees the flag finds
        // the cell.
        self.broadcasting
            .store(true, std::sync::atomic::Ordering::Release);
        cell
    }

    /// The broadcast cell for the output at `slot`, if one was asked
    /// for. `None` is the common answer: a program nobody composed
    /// under has none at all, which the caller checks first.
    #[inline]
    pub(crate) fn published_output(&self, slot: usize) -> Option<crate::kernel::SharedCell> {
        let cells = self
            .scope
            .output_cells
            .lock()
            .expect("output cells poisoned");
        cells.get(slot)?.clone()
    }

    /// Whether any descendant asked for a broadcast cell: one lock-free
    /// check on every pull, by name or by index, false for every program
    /// with no subscope, and what keeps `published_output`'s lock off
    /// the pull path until a descendant exists.
    #[inline]
    pub(crate) fn broadcasts(&self) -> bool {
        self.broadcasting.load(std::sync::atomic::Ordering::Acquire)
    }

    /// The next bit of the intent word. The compiled kernel keeps one
    /// word, so cells past the 64th share bit 63, where the interpreter
    /// opens a new word (`allocate_cell_bit`).
    fn next_cell_bit(&self) -> u8 {
        use std::sync::atomic::Ordering;
        let mut bit = self.next_bit.load(Ordering::Relaxed);
        loop {
            let next = bit.saturating_add(1).min(63);
            match self.next_bit.compare_exchange_weak(
                bit,
                next,
                Ordering::Relaxed,
                Ordering::Relaxed,
            ) {
                Ok(_) => return bit.min(63),
                Err(seen) => bit = seen,
            }
        }
    }

    /// A cell of this kernel's scope holding `initial`, with the next
    /// bit of the intent word.
    fn new_cell(&self, initial: Value) -> crate::kernel::SharedCell {
        std::sync::Arc::new(crate::kernel::SharedCellInner::new(
            initial,
            self.intent.clone(),
            self.next_cell_bit(),
        ))
    }

    /// Give every `shared` slot a cell of its own holding its current
    /// value: what a kernel created from a shared program starts with,
    /// as an interpreter state seeds its own cells.
    pub(crate) fn reseed_cells(&mut self) {
        self.intent = std::sync::Arc::new(std::sync::atomic::AtomicU64::new(0));
        self.next_bit.store(0, std::sync::atomic::Ordering::Relaxed);
        // A new state publishes to nobody: the descendants bound to the
        // state this one came from are not bound to this one.
        self.scope
            .output_cells
            .lock()
            .expect("output cells poisoned")
            .clear();
        for i in 0..self.slots.len() {
            if self.slots[i].cell.is_none() {
                continue;
            }
            let cell = self.new_cell(self.slots[i].value.clone());
            self.slots[i].seen = Some(cell.snapshot().1);
            self.slots[i].cell = Some(cell);
        }
    }

    /// Bind the `shared` binding `name` to `cell`: from now on this
    /// kernel reads and writes that register, as every other holder of
    /// the cell does. Returns the slot, for the caller's dirty marking;
    /// the value arrives at the next refresh.
    pub(crate) fn attach_cell(
        &mut self,
        name: &str,
        cell: crate::kernel::SharedCell,
    ) -> Result<usize, String> {
        let Some(&i) = self.by_name.get(name) else {
            let known: Vec<&str> = self
                .slots
                .iter()
                .filter(|s| s.cell.is_some())
                .map(|s| s.name.as_str())
                .collect();
            return Err(format!(
                "no `shared` binding named '{name}'; this kernel's shared bindings are {known:?}"
            ));
        };
        let s = &mut self.slots[i];
        if s.cell.is_none() {
            return Err(format!(
                "'{name}' is an extern, not a `shared` binding; only a `shared` binding takes a cell"
            ));
        }
        s.cell = Some(cell);
        s.seen = None;
        Ok(s.slot)
    }

    /// Bind an input slot to `cell` whether or not it was built with
    /// one, and answer the slot. `None` when this kernel has no extern
    /// by that name.
    ///
    /// This is the binder's attach, not a host's. A parent wiring a
    /// child attaches its cells to whichever of the child's input
    /// slots match by name — the child declared them `extern`, and it
    /// is the parent that decides one of them reads a register — where
    /// [`Self::attach_cell`] is a host joining two kernels at a slot
    /// that is already a register on both, and refuses the rest. The
    /// interpreter has had both all along: its `Kernel` impl filters to
    /// `shared` outputs and its binder calls the state directly.
    ///
    /// `seen` is cleared, so the next refresh takes the cell's value.
    pub(crate) fn bind_cell(
        &mut self,
        name: &str,
        cell: crate::kernel::SharedCell,
    ) -> Option<usize> {
        let &i = self.by_name.get(name)?;
        let s = &mut self.slots[i];
        s.cell = Some(cell);
        s.seen = None;
        Some(s.slot)
    }

    /// The cells this kernel's `shared` bindings are bound to.
    pub(crate) fn shared_cells(&self) -> Vec<crate::kernel::SharedCellEntry> {
        self.slots
            .iter()
            .filter_map(|s| {
                s.cell.as_ref().map(|cell| crate::kernel::SharedCellEntry {
                    name: s.name.clone(),
                    port_type: s.ty,
                    cell: cell.clone(),
                })
            })
            .collect()
    }

    /// Whether any cell has been published to since this kernel last
    /// took its value: one Acquire load per cell.
    pub(crate) fn cells_dirty(&self) -> bool {
        self.slots.iter().any(|s| match (&s.cell, s.seen) {
            (Some(cell), seen) => {
                Some(cell.revision.load(std::sync::atomic::Ordering::Acquire)) != seen
            }
            (None, _) => false,
        })
    }

    /// Take every cell's current value where its revision moved since
    /// this kernel last read it: the value is stored and written
    /// through, and the slot is recorded for the kernel to mark dirty
    /// (`take_changed`).
    pub(crate) fn refresh_cells(&mut self, buffer: &mut [u64]) {
        for s in &mut self.slots {
            let Some(cell) = &s.cell else {
                continue;
            };
            if Some(cell.revision.load(std::sync::atomic::Ordering::Acquire)) == s.seen {
                continue;
            }
            let (value, revision) = cell.snapshot();
            let was_counted = s.counted_unset();
            s.value = value;
            s.seen = Some(revision);
            // Revision 0 is the value the cell was made with; any later
            // one is a write, which seeds a register.
            if revision > 0 {
                s.seeded();
            }
            self.unset_read = track_unset(self.unset_read, was_counted, s);
            write_through(s, buffer);
            self.changed.push(s.slot);
        }
    }

    /// Whether the last refresh changed any slot.
    #[inline]
    pub(crate) fn has_changed(&self) -> bool {
        !self.changed.is_empty()
    }

    /// The slots the last refresh changed, once.
    pub(crate) fn take_changed(&mut self) -> Vec<usize> {
        std::mem::take(&mut self.changed)
    }

    /// Give back the drained list so its allocation is reused.
    pub(crate) fn return_changed(&mut self, mut list: Vec<usize>) {
        list.clear();
        self.changed = list;
    }

    /// Every input by name, the coordinates first.
    pub(crate) fn input_names(&self) -> &[String] {
        &self.input_names
    }

    /// Record the named outputs in declaration order.
    pub(crate) fn set_output_names(&mut self, names: &[String]) {
        self.output_names = names.to_vec();
    }

    /// The binding modifiers of the named outputs, as the assembler
    /// resolved them. A compiled kernel keeps no `PolydatProgram`, so
    /// without these it cannot answer whether an output is `const` —
    /// which a binder asks, to send a `const` output down the
    /// value-copy path rather than attach a cell to it
    /// ([scope_model.md](scope_model.md) §4).
    pub(crate) fn set_output_modifiers(
        &mut self,
        modifiers: &HashMap<String, crate::dsl::ast::BindingModifier>,
    ) {
        self.scope.output_modifiers = modifiers.clone();
    }

    /// Record the const bindings a kernel initializes.
    /// Marks each const's slot and fallback input as not counted unset,
    /// as only initialization reads them, and each `shared` register
    /// initialization seeds as not counted until it is seeded.
    pub(crate) fn set_const_inits(&mut self, inits: &[crate::kernel::ConstInit]) {
        self.scope.const_inits = inits.to_vec();
        for s in &mut self.slots {
            s.register_start = register_start(&s.name, inits);
            s.init_only = init_only(&s.name, s.is_const, inits);
        }
        self.recount_unset();
    }

    /// Count the unset externs native code reads, from scratch.
    fn recount_unset(&mut self) {
        self.unset_read = self.slots.iter().filter(|s| s.counted_unset()).count() as u32;
    }

    /// The const bindings a kernel initializes, in dependency order.
    pub(crate) fn const_inits(&self) -> &[crate::kernel::ConstInit] {
        &self.scope.const_inits
    }

    /// Record the outputs whose value is fixed for a kernel's life.
    pub(crate) fn set_fixed_outputs(&mut self, names: std::collections::HashSet<String>) {
        self.scope.fixed_outputs = names;
    }

    /// Whether the output's value is fixed for a kernel's life.
    pub(crate) fn is_fixed_output(&self, name: &str) -> bool {
        self.scope.fixed_outputs.contains(name)
    }

    /// Whether the input at `index` holds a const's value, which only
    /// initialization writes.
    pub(crate) fn is_const_index(&self, index: usize) -> bool {
        matches!(self.by_index.get(index), Some(Some(i)) if self.slots[*i].is_const)
    }

    /// [`Self::is_const_index`] by name.
    pub(crate) fn is_const_name(&self, name: &str) -> bool {
        self.by_name
            .get(name)
            .is_some_and(|&i| self.slots[i].is_const)
    }

    /// The declared type of a named input slot. A coordinate has none
    /// here — it is not an extern — and answers `U64`, which is what
    /// every coordinate is.
    pub(crate) fn input_port_type(&self, name: &str) -> Option<PortType> {
        if let Some(&i) = self.by_name.get(name) {
            return Some(self.slots[i].reported);
        }
        self.input_names
            .iter()
            .any(|n| n == name)
            .then_some(PortType::U64)
    }

    /// How input `name`'s type was established (input_variance.md §3).
    pub(crate) fn input_type_origin(&self, name: &str) -> Option<crate::kernel::TypeOrigin> {
        let i = self.input_names.iter().position(|n| n == name)?;
        self.scope.origins.get(i).copied()
    }

    /// The binding modifier of a named output; `NONE` for a name this
    /// kernel does not declare, as the interpreter's program answers.
    pub(crate) fn output_modifier(&self, name: &str) -> crate::dsl::ast::BindingModifier {
        self.scope
            .output_modifiers
            .get(name)
            .copied()
            .unwrap_or(crate::dsl::ast::BindingModifier::NONE)
    }

    /// Cells this kernel carries forward for a descendant without
    /// holding a slot for them itself: what the interpreter calls
    /// transit. A compiled kernel forwards them the same way, so a
    /// grandchild binds to a `shared` cell its parent's program never
    /// named.
    pub(crate) fn set_transit_cells(&mut self, cells: Vec<crate::kernel::SharedCellEntry>) {
        self.scope.transit_cells = cells;
    }

    /// The cells this kernel's own `shared` slots hold, plus the ones
    /// it carries forward: every cell a descendant could bind to,
    /// which is what "in scope" means.
    pub(crate) fn cells_in_scope(&self) -> Vec<crate::kernel::SharedCellEntry> {
        let mut by_name: HashMap<String, crate::kernel::SharedCellEntry> = self
            .scope
            .transit_cells
            .iter()
            .map(|e| (e.name.clone(), e.clone()))
            .collect();
        for entry in self.shared_cells() {
            by_name.insert(entry.name.clone(), entry);
        }
        by_name.into_values().collect()
    }

    /// Every named output in declaration order.
    pub(crate) fn output_names(&self) -> &[String] {
        &self.output_names
    }

    /// The slots of the externs that have no value at build: they are
    /// `None` until a host sets them, and their consumers propagate it.
    #[cfg(feature = "jit")]
    pub(crate) fn unset_slots(&self) -> Vec<usize> {
        self.slots
            .iter()
            .filter(|s| s.value == Value::None)
            .map(|s| s.slot)
            .collect()
    }

    /// Every buffer slot of the externs native code reads that a host
    /// can leave with no value, sorted: each extern but a const's slot
    /// and a const's fallback input, with a `shared` register that has a
    /// computed start among them, as it counts once seeded. What the
    /// fusion units split by (`fusion_units::refine_by_externs`).
    #[cfg(feature = "jit")]
    pub(crate) fn unset_read_slots(&self) -> Vec<usize> {
        let mut slots: Vec<usize> = self
            .slots
            .iter()
            .filter(|s| !s.init_only || s.register_start)
            .flat_map(|s| {
                let pair = s.ty.slot_color() == crate::ast::SlotColor::Ref2;
                std::iter::once(s.slot).chain(pair.then_some(s.slot + 1))
            })
            .collect();
        slots.sort_unstable();
        slots
    }

    /// The cursors the program declares, with their partitions where
    /// the compiler resolved them.
    pub(crate) fn cursor_schemas(&self) -> &[crate::iteration::source::SourceSchema] {
        &self.cursors
    }

    /// Record the extent of cursor `index`, resolved after the build from
    /// the constants the graph folded.
    pub(crate) fn set_cursor_extent(&mut self, index: usize, extent: u64) {
        if let Some(schema) = self.cursors.get_mut(index) {
            schema.extent = Some(extent);
        }
    }

    /// The writes that narrow cursor `name` to `partition`: its `Ext`
    /// slot and its six scalar projections, each an extern of this
    /// kernel. An unknown cursor is an error naming the known ones.
    pub(crate) fn cursor_writes(
        &self,
        name: &str,
        partition: &crate::iteration::cursor_partition::Partition,
    ) -> Result<Vec<(String, Value)>, WriteError> {
        if !self.cursors.iter().any(|c| c.name == name) {
            return Err(WriteError::UnknownWire {
                key: name.to_string(),
                known: self.cursors.iter().map(|c| c.name.clone()).collect(),
            });
        }
        Ok(
            crate::iteration::cursor_partition::cursor_slot_writes(name, partition)
                .into_iter()
                .filter(|(slot, _)| self.by_name.contains_key(slot))
                .collect(),
        )
    }

    /// Write every extern into the buffer, and mark the unset ones in
    /// `none` where the kernel keeps a mask; returns whether any is
    /// unset. What a build and a reset do.
    pub(crate) fn seed(&self, buffer: &mut [u64], mut none: Option<&mut [bool]>) -> bool {
        let mut any_none = false;
        for s in &self.slots {
            write_through(s, buffer);
            let unset = s.value == Value::None;
            any_none |= unset;
            if let Some(mask) = none.as_deref_mut() {
                mask[s.slot] = unset;
            }
        }
        any_none
    }

    /// Whether the extern whose first buffer slot is `slot` has no value.
    pub(crate) fn slot_is_unset(&self, slot: usize) -> bool {
        self.slots
            .iter()
            .any(|s| s.slot == slot && s.value == Value::None)
    }

    /// Whether any extern has no value: a kernel that keeps a `None`
    /// mask reads the mask only then, and propagates the `None` as the
    /// interpreter does (engines.md §3.3). Pure native, which keeps no
    /// mask, uses `any_unset_read` and refuses only the pulls that
    /// depend on an unset extern.
    pub(crate) fn any_unset(&self) -> bool {
        self.slots.iter().any(|s| s.value == Value::None)
    }

    /// Whether an extern native code reads has no value: one load, what
    /// a pure native run checks before it runs.
    #[cfg(feature = "jit")]
    #[inline]
    pub(crate) fn any_unset_read(&self) -> bool {
        self.unset_read != 0
    }

    /// The name and type of the first extern native code reads that has
    /// no value and one of whose buffer slots `reads` accepts, for a
    /// native kernel's refusal. A const's slot and its fallback input
    /// are not among them: only initialization reads or writes those,
    /// from Rust, and every reader of the const reads the passthrough of
    /// its slot once it is set. Nor is a `shared` register with a
    /// computed start before initialization seeds it.
    #[cfg(feature = "jit")]
    pub(crate) fn first_unset(&self, reads: impl Fn(usize) -> bool) -> Option<(&str, PortType)> {
        self.slots
            .iter()
            .find(|s| {
                s.counted_unset()
                    && (reads(s.slot)
                        || (s.ty.slot_color() == crate::ast::SlotColor::Ref2 && reads(s.slot + 1)))
            })
            .map(|s| (s.name.as_str(), s.ty))
    }

    /// Set an extern by name. The value must be of the declared port
    /// type; `Value::None` clears it to unset. The slot is written
    /// through now. Returns the extern's first slot, for the caller's
    /// dirty marking, and whether it is now unset.
    pub(crate) fn set(
        &mut self,
        name: &str,
        value: Value,
        buffer: &mut [u64],
    ) -> Result<(usize, bool), WriteError> {
        let Some(&i) = self.by_name.get(name) else {
            if self.input_names.iter().any(|n| n == name) {
                return Err(WriteError::CoordinateSlot {
                    slot: name.to_string(),
                });
            }
            // Every input slot this kernel has, coordinates included,
            // as `Kernel::input_names` reports them and as the indexed
            // write beside this one already listed. A caller who
            // mistyped a coordinate meant a name that exists, and is
            // not helped by a list that omits it; the exact match is
            // the `CoordinateSlot` above.
            return Err(WriteError::UnknownWire {
                key: name.to_string(),
                known: self.input_names.clone(),
            });
        };
        self.set_slot(i, value, buffer)
    }

    /// [`Self::set`] by input index, the index among every input with
    /// the coordinates first, as `input_names` lists them.
    pub(crate) fn set_at(
        &mut self,
        index: usize,
        value: Value,
        buffer: &mut [u64],
    ) -> Result<(usize, bool), WriteError> {
        match self.by_index.get(index) {
            Some(Some(i)) => self.set_slot(*i, value, buffer),
            Some(None) => Err(WriteError::CoordinateSlot {
                slot: self.input_names[index].clone(),
            }),
            None => Err(WriteError::UnknownWire {
                key: format!("wire[{index}]"),
                known: self.input_names.clone(),
            }),
        }
    }

    /// The one write rule of every engine: the value satisfies the
    /// declared type (a carrier's bit-stuffed forms included) or is
    /// `None`, which clears the extern.
    fn set_slot(
        &mut self,
        i: usize,
        value: Value,
        buffer: &mut [u64],
    ) -> Result<(usize, bool), WriteError> {
        let s = &mut self.slots[i];
        if !value.satisfies_slot(s.ty) {
            return Err(WriteError::TypeMismatch {
                slot: s.name.clone(),
                expected: s.ty,
                got: value.port_type(),
            });
        }
        let was_counted = s.counted_unset();
        s.value = value;
        // A `shared` binding's slot writes through its cell, so every
        // holder of the cell reads this value; the revision is this
        // kernel's own and needs no refresh.
        if let Some(cell) = &s.cell {
            cell.publish(s.value.clone());
            s.seen = Some(cell.revision.load(std::sync::atomic::Ordering::Acquire));
        }
        s.seeded();
        write_through(s, buffer);
        let s = &self.slots[i];
        self.unset_read = track_unset(self.unset_read, was_counted, s);
        Ok((s.slot, s.value == Value::None))
    }

    /// Start over from the program: every extern back at its declared
    /// default, written through into `buffer`, and every `shared`
    /// binding with a cell of its own holding that default. What a
    /// kernel created from a shared program starts with, whatever the
    /// kernel it was cloned from had been set to. Also what a clone
    /// needs before its first run: its pairs must point into its own
    /// stored values, not the original's.
    pub(crate) fn reset_to_program(&mut self, buffer: &mut [u64]) {
        for s in &mut self.slots {
            s.value = s.default.clone();
            s.seen = None;
            // The register's new cell is unwritten until initialization
            // seeds it again.
            if s.register_start {
                s.init_only = true;
            }
            write_through(s, buffer);
        }
        self.recount_unset();
        self.reseed_cells();
    }

    /// The current value of the extern `name`, if there is one.
    pub(crate) fn value(&self, name: &str) -> Option<Value> {
        self.by_name.get(name).map(|&i| self.slots[i].value.clone())
    }

    /// How many inputs are coordinates: the leading inputs with no
    /// extern slot. Not the core's `coord_count`, which counts the
    /// buffer slots every input occupies.
    #[inline]
    pub(crate) fn coordinate_count(&self) -> usize {
        self.coordinates as usize
    }

    /// The extern at input `index`, `None` for a coordinate or past the
    /// end.
    fn slot_at(&self, index: usize) -> Option<&ExternSlot> {
        self.by_index
            .get(index)
            .copied()
            .flatten()
            .map(|i| &self.slots[i])
    }

    /// The current value of the extern at input `index`.
    pub(crate) fn value_at(&self, index: usize) -> Option<Value> {
        // A cell-bound slot's value is the cell's; the slot holds the
        // copy taken at the last refresh, which a pull makes.
        self.slot_at(index).map(|s| match &s.cell {
            Some(cell) => cell.snapshot().0,
            None => s.value.clone(),
        })
    }

    /// The declared default of the extern at input `index`.
    pub(crate) fn default_at(&self, index: usize) -> Option<Value> {
        self.slot_at(index).map(|s| s.default.clone())
    }

    /// Whether the extern at input `index` is bound to a shared cell.
    pub(crate) fn is_cell_bound_at(&self, index: usize) -> bool {
        self.slot_at(index).is_some_and(|s| s.cell.is_some())
    }

    /// The Rule 2 write-throughs this kernel commits.
    pub(crate) fn write_throughs(&self) -> &[(String, String)] {
        &self.scope.write_throughs
    }

    /// Record the Rule 2 write-throughs this kernel commits.
    pub(crate) fn set_write_throughs(&mut self, pairs: Vec<(String, String)>) {
        self.scope.write_throughs = pairs;
    }

    /// The externs by name and declared type, for diagnostics.
    pub(crate) fn names(&self) -> Vec<(&str, PortType)> {
        self.slots
            .iter()
            .map(|s| (s.name.as_str(), s.reported))
            .collect()
    }
}

/// `count`, the unset externs native code reads, after `s` changed from
/// counted (`was_counted`) or not to what it is now.
#[inline]
fn track_unset(count: u32, was_counted: bool, s: &ExternSlot) -> u32 {
    match (was_counted, s.counted_unset()) {
        (false, true) => count + 1,
        (true, false) => count - 1,
        _ => count,
    }
}

/// Write an extern's current value into its slots: a carrier as its
/// bits, a `Ref2` kind as the pair into the value the slot stores. An
/// unset `Ref2` kind is an empty pair, which a string consumer reads
/// as empty where nothing keeps a `None` mask and a value consumer
/// reads as `None`.
fn write_through(s: &ExternSlot, buffer: &mut [u64]) {
    match s.ty.slot_color() {
        crate::ast::SlotColor::Ref2 => {
            let (p, l) = match &s.value {
                Value::None => crate::compile::marshal::empty_pair(),
                // A `Dyn` slot names the stored value itself, whatever
                // its variant, for its converter to read.
                v if s.ty == PortType::Dyn => (v as *const Value as usize as u64, 1),
                v => crate::compile::marshal::borrow_pair(v).unwrap_or_else(|| {
                    panic!(
                        "extern '{}' ({}) holds a {} value, which has no slot form",
                        s.name,
                        s.ty,
                        v.port_type()
                    )
                }),
            };
            buffer[s.slot] = p;
            buffer[s.slot + 1] = l;
        }
        _ => buffer[s.slot] = carrier_bits(&s.value),
    }
}

/// A carrier value's slot bits; an unset carrier reads as zero.
///
/// `None` is the only case that means zero. Anything else reaching the
/// catch-all is a slot coloured `Imm1` holding something that does not
/// fit one, which the type system does not allow, so it panics rather
/// than answer a zero with no error.
fn carrier_bits(v: &Value) -> u64 {
    match v {
        Value::U64(n) => *n,
        Value::I64(n) => *n as u64,
        Value::F64(f) => f.to_bits(),
        Value::Bool(b) => u64::from(*b),
        Value::None => 0,
        other => panic!(
            "an extern slot holds a {:?}, which is not a one-slot carrier",
            other.port_type()
        ),
    }
}