polydat_core/kernel/engines.rs
1// Copyright 2024-2026 Jonathan Shook
2// SPDX-License-Identifier: Apache-2.0
3
4//! Polydat evaluation engines: EngineCore (shared eval loop) and the three
5//! P1 engine types — PolydatState (dependent-list), RawState (no provenance),
6//! and ProvScanState (provenance-scan).
7
8use std::sync::atomic::{AtomicU64, Ordering};
9use std::sync::{Arc, Mutex, OnceLock};
10
11use super::WireSource;
12use super::program::PolydatProgram;
13use crate::ast::Value;
14
15/// Cached lookup of the `NBRS_DIRTY_DEBUG` env var. Called from
16/// the per-cycle hot path (`PolydatState::set_input`); reading the
17/// real `std::env::var` on every cycle costs ~30% of CPU on
18/// single-fiber dryrun benches (it walks the libc env table and
19/// formats a fresh CString each call). The OnceLock evaluates
20/// once on first touch and every subsequent call is one atomic
21/// load.
22fn nbrs_dirty_debug_enabled() -> bool {
23 static FLAG: OnceLock<bool> = OnceLock::new();
24 *FLAG.get_or_init(|| std::env::var("NBRS_DIRTY_DEBUG").is_ok())
25}
26
27/// A cross-kernel mutable cell for a `shared`-modifier wire.
28///
29/// When a `shared` output in an outer scope is bound into an
30/// inner kernel via `materialize_wiring_from_outer`, both kernels' input
31/// slots reference the same `SharedCell`. Writes from inner via
32/// `set_input` flow through to the cell; reads on either side
33/// pick up the latest value.
34///
35/// Concurrent writers serialize at the Mutex; the current
36/// semantic is **last-write-wins** (lock-acquisition order).
37/// Future templated patterns (atomic-fetch-add, sum-reduction,
38/// merge, etc.) — see SRD-16 §"Open: concurrent shared
39/// mutation" — will introduce alternative cell types selected
40/// per binding declaration.
41///
42/// ## Cross-fiber validity tracking
43///
44/// Each cell carries its own validity-tracking handles per
45/// `polydat/docs/design/cross_fiber_invalidation.md`:
46///
47/// - `revision: AtomicU64` — monotonic counter, bumped on every
48/// write. Consumer fibers cache the last revision they
49/// observed in their per-fiber `last_seen` map; a mismatch
50/// tells the cone walker to re-evaluate.
51/// - `scope_intent_dirty: Arc<AtomicU64>` — one intent word, shared
52/// with every other cell allocated from the same word. The
53/// cell's `bit` position is set on every write, allowing
54/// consumers to do an O(1) bulk check ("any cell in this
55/// scope dirty?") before drilling down to the per-cell
56/// revision compare.
57/// - `bit: u8` — this cell's position within its word. The scope
58/// keeps one `Arc<AtomicU64>` per 64 cells, grown on demand by
59/// the defining scope's `EngineCore::allocate_cell_bit`.
60///
61/// The reader contract (S5 §1.1) is preserved: a producer's
62/// `publish` writes value + revision + intent bit in three
63/// Release stores; a consumer's `check_clean` walk on its next
64/// read observes the change without any host-side ceremony.
65pub struct SharedCellInner {
66 /// Cell value. The mutex serialises concurrent writers and
67 /// gives readers single-value atomicity.
68 pub value: Mutex<Value>,
69 /// Monotonic revision counter. Bumped on every write
70 /// (Release); compared by consumers (Acquire) against
71 /// per-fiber `last_seen`.
72 pub revision: AtomicU64,
73 /// Defining scope's intent-dirty bit-vector. Shared by Arc
74 /// across every cell allocated by the same scope. On every
75 /// write the producer ORs `1 << self.bit` into this
76 /// (Release) so consumers' bulk-mask check sees the scope
77 /// as dirty.
78 pub scope_intent_dirty: Arc<AtomicU64>,
79 /// This cell's bit position in `scope_intent_dirty`. Stable
80 /// for the cell's lifetime.
81 pub bit: u8,
82}
83
84impl std::fmt::Debug for SharedCellInner {
85 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
86 f.debug_struct("SharedCellInner")
87 .field("revision", &self.revision.load(Ordering::Relaxed))
88 .field("bit", &self.bit)
89 .finish_non_exhaustive()
90 }
91}
92
93impl SharedCellInner {
94 /// Construct a new cell with the given initial value, bound
95 /// to the defining scope's intent-dirty word at the given
96 /// bit position within that word. Callers must allocate
97 /// `(word, bit)` via `EngineCore::allocate_cell_bit` —
98 /// the bit is not reusable for the cell's lifetime.
99 pub fn new(initial: Value, scope_intent_dirty: Arc<AtomicU64>, bit: u8) -> Self {
100 debug_assert!(
101 bit < 64,
102 "bit-within-word {bit} must be < 64; the allocator splits >64-bit \
103 scope vectors across multiple words"
104 );
105 Self {
106 value: Mutex::new(initial),
107 revision: AtomicU64::new(0),
108 scope_intent_dirty,
109 bit,
110 }
111 }
112
113 /// Producer-side write: replace the cell's value, bump the
114 /// revision, set the intent bit. Three Release stores
115 /// publish the write across any consumer fiber per
116 /// `cross_fiber_invalidation.md` §6. The mutex critical
117 /// section is held only for the value swap; the atomics
118 /// run outside it.
119 pub fn publish(&self, value: Value) {
120 {
121 let mut guard = self.value.lock().unwrap();
122 *guard = value;
123 }
124 self.revision.fetch_add(1, Ordering::Release);
125 self.scope_intent_dirty
126 .fetch_or(1u64 << self.bit, Ordering::Release);
127 }
128
129 /// Consumer-side read: snapshot the cell's value and the
130 /// revision it was published at. Returns a pair so the
131 /// caller can update its `last_seen[cell] = revision`
132 /// alongside taking the value, without a second cell access.
133 pub fn snapshot(&self) -> (Value, u64) {
134 // Acquire-load the revision first so the value read
135 // synchronises-with the producer's value publication.
136 // The mutex itself provides the memory barrier for the
137 // value, but the revision is read with explicit Acquire
138 // for the cross-fiber happens-before relation.
139 let value = self.value.lock().unwrap().clone();
140 let revision = self.revision.load(Ordering::Acquire);
141 (value, revision)
142 }
143}
144
145/// Externally-held handle to a shared cell. `Arc<SharedCellInner>`
146/// so a single cell can be referenced from many kernels at
147/// once. The handle is cheap to clone (Arc bump).
148pub type SharedCell = Arc<SharedCellInner>;
149
150/// Per-node cone metadata for cell-bound input dependencies.
151///
152/// Built lazily on first `check_cell_clean` per node and
153/// cached in [`EngineCore::cell_cones`]; invalidated by
154/// clearing the cache whenever cells are attached or detached.
155///
156/// The structure groups a node's cell-bound input dependencies
157/// by the defining scope's `intent_dirty` Arc (compared by
158/// `Arc::ptr_eq`). Each group carries the bulk-check
159/// `interest_mask` for the scope plus per-cell drill-down
160/// entries — implementing the bulk-mask + per-cell-revision
161/// protocol from `cross_fiber_invalidation.md` §5.
162#[derive(Debug, Default, Clone)]
163pub(crate) struct CellCone {
164 /// Cells grouped by defining-scope's `intent_dirty`. Empty
165 /// = no cell-bound deps; check returns trivially clean.
166 pub(crate) groups: Vec<CellConeGroup>,
167}
168
169#[derive(Debug, Clone)]
170pub(crate) struct CellConeGroup {
171 /// Defining scope's `intent_dirty` vector (Arc cloned from
172 /// the cells). Bulk-mask check: AND this against
173 /// `interest_mask`; if zero, every cell in this group is
174 /// clean for this consumer (modulo last_seen) — skip the
175 /// drill-down.
176 pub(crate) intent_dirty: Arc<AtomicU64>,
177 /// OR of `1 << cell.bit` for every cell in this group.
178 pub(crate) interest_mask: u64,
179 /// Per-cell drill-down entries. Each gives the bit position
180 /// in `intent_dirty` plus the input slot index where the
181 /// cell is attached, for revision compare against
182 /// `last_seen`.
183 pub(crate) cells: Vec<CellConeEntry>,
184}
185
186#[derive(Debug, Clone, Copy)]
187pub(crate) struct CellConeEntry {
188 pub(crate) bit: u8,
189 pub(crate) input_slot: usize,
190}
191
192/// One named shared cell propagated through the parent → child
193/// scope chain. Carried on `PolydatKernel` (and surfaced through
194/// `ScopeKernel::shared_cells_in_scope`) so a descendant whose
195/// program declares a matching input slot can attach the cell —
196/// even when intermediate scopes' bodies never name it and so
197/// have no input slot for it themselves.
198///
199/// Without this carrier, an ancestral `shared X := …` cell
200/// becomes invisible past the first intermediate scope under
201/// the closure-binding economy. With it, every spawn step
202/// computes "every cell visible at this scope" and threads the
203/// full set forward — the cascade is transitive by
204/// construction.
205#[derive(Clone, Debug)]
206pub struct SharedCellEntry {
207 /// The binding's name.
208 pub name: String,
209 /// The cell's declared type.
210 pub port_type: crate::ast::PortType,
211 /// The cell.
212 pub cell: SharedCell,
213}
214
215/// SRD-82 §"Panic reporting: one full render" — set by a host
216/// runtime that catches worker panics and renders the full
217/// enriched diagnostic itself (the `errors:` block). When set,
218/// the re-raise hook below prints a single first-line notice
219/// instead of the full body; bare polydat consumers never set it
220/// and keep the full print.
221static PANIC_REPORTING_DOWNSTREAM: std::sync::atomic::AtomicBool =
222 std::sync::atomic::AtomicBool::new(false);
223
224/// Declare that a downstream reporter will render eval-panic
225/// diagnostics in full (see `PANIC_REPORTING_DOWNSTREAM`).
226pub fn set_panic_reporting_downstream(on: bool) {
227 PANIC_REPORTING_DOWNSTREAM.store(on, std::sync::atomic::Ordering::Relaxed);
228}
229
230thread_local! {
231 /// True while a node eval runs inside the enrichment
232 /// catch_unwind in `eval_node`. The suppression hook checks
233 /// this to swallow the raw std panic-hook print (bare payload
234 /// + backtrace pointer at the original panic site) — that
235 /// same panic is about to be caught, enriched with
236 /// node/output/input context, and re-raised via `panic_any`,
237 /// which fires the hook again with this flag clear. Net
238 /// effect: exactly ONE hook print, and it's the enriched one.
239 static EVAL_PANIC_CAPTURE: std::cell::Cell<bool> =
240 const { std::cell::Cell::new(false) };
241 /// Original panic location captured by the suppression hook
242 /// while the flag above is set. Folded into the enriched
243 /// message so the true `file:line` survives the re-raise
244 /// (the re-raised panic's own location points at the
245 /// re-raise site, which is useless).
246 static EVAL_PANIC_LOCATION: std::cell::RefCell<Option<String>> =
247 const { std::cell::RefCell::new(None) };
248 /// One-shot marker armed just before the enriched re-raise
249 /// when a downstream reporter exists: the hook prints a short
250 /// first-line notice for that panic instead of the full body.
251 static RERAISE_SHORT: std::cell::Cell<bool> =
252 const { std::cell::Cell::new(false) };
253}
254
255/// Install (once, process-wide) a panic hook that chains to the
256/// previously installed hook unless the current thread is inside
257/// the wrapped node eval, in which case it records the panic
258/// location and stays quiet.
259fn install_eval_panic_hook() {
260 static HOOK: std::sync::Once = std::sync::Once::new();
261 HOOK.call_once(|| {
262 let prev = std::panic::take_hook();
263 std::panic::set_hook(Box::new(move |info| {
264 if EVAL_PANIC_CAPTURE.with(|c| c.get()) {
265 let loc = info.location().map(|l| l.to_string());
266 EVAL_PANIC_LOCATION.with(|slot| *slot.borrow_mut() = loc);
267 } else if RERAISE_SHORT.with(|c| c.replace(false)) {
268 // The runtime will render the full enriched
269 // diagnostic in the phase error list; one short
270 // line keeps the terminal signal without the
271 // four-fold repeat (SRD-82 §one full render).
272 let first = info
273 .payload()
274 .downcast_ref::<String>()
275 .map(String::as_str)
276 .and_then(|m| m.lines().next())
277 .unwrap_or("<non-string panic payload>");
278 eprintln!("op eval panic (detail in phase errors): {first}");
279 } else {
280 prev(info);
281 }
282 }));
283 });
284}
285
286/// RAII guard arming the suppression hook for one wrapped eval.
287/// Saves and restores the previous flag value: nodes that drive
288/// sub-kernels (comprehensions, gk-call) nest evals, and each
289/// level's catch_unwind must see its own panics suppressed.
290pub(crate) struct EvalPanicCaptureGuard {
291 prev: bool,
292}
293
294impl EvalPanicCaptureGuard {
295 pub(crate) fn arm() -> Self {
296 install_eval_panic_hook();
297 let prev = EVAL_PANIC_CAPTURE.with(|c| c.replace(true));
298 EVAL_PANIC_LOCATION.with(|slot| slot.borrow_mut().take());
299 Self { prev }
300 }
301}
302
303impl Drop for EvalPanicCaptureGuard {
304 fn drop(&mut self) {
305 EVAL_PANIC_CAPTURE.with(|c| c.set(self.prev));
306 }
307}
308
309/// The text of a panic payload: a `String` or a `&str`, else a marker.
310pub(crate) fn panic_payload_text(payload: &(dyn std::any::Any + Send)) -> String {
311 payload
312 .downcast_ref::<&'static str>()
313 .map(|s| (*s).to_string())
314 .or_else(|| payload.downcast_ref::<String>().cloned())
315 .unwrap_or_else(|| "<non-string panic payload>".into())
316}
317
318/// Build the rich diagnostic message for a node-level eval panic, on
319/// every engine (engine_parity.md, A7): the original payload, the
320/// panic location the capture guard recorded, the node's function
321/// name, every output it feeds, the program's diagnostic context
322/// (typically the source path / scope label), and the input values,
323/// already formatted (the interpreter's `Value`s through
324/// [`format_value_for_diag`], a compiled kernel's slots through its
325/// decoder). This is what the user sees instead of the bare panic
326/// payload, and it reads the same whichever engine raised it.
327pub(crate) fn enrich_panic(
328 payload: Box<dyn std::any::Any + Send>,
329 node_name: &str,
330 output_names: &[&str],
331 context: &str,
332 inputs: &[String],
333) -> String {
334 let original = panic_payload_text(payload.as_ref());
335 // A payload that already carries node context came from a
336 // nested wrapped eval's re-raise; its captured "location" is
337 // the re-raise site, not the original panic — skip it.
338 let location_line = if original.contains("↳ in node") {
339 String::new()
340 } else {
341 EVAL_PANIC_LOCATION
342 .with(|slot| slot.borrow_mut().take())
343 .map(|loc| format!("\n ↳ panicked at {loc}"))
344 .unwrap_or_default()
345 };
346 let outputs_label = if output_names.is_empty() {
347 "no declared output".to_string()
348 } else {
349 format!(
350 "output{} {}",
351 if output_names.len() == 1 { "" } else { "s" },
352 output_names.join(", ")
353 )
354 };
355 let mut input_label = String::new();
356 for (i, v) in inputs.iter().enumerate() {
357 if i > 0 {
358 input_label.push_str(", ");
359 }
360 input_label.push_str(&format!("[{i}]={v}"));
361 }
362 format!(
363 "{original}{location_line}\n ↳ in node `{node_name}` ({outputs_label}) \
364 while evaluating {context}\n \
365 ↳ inputs: [{input_label}]"
366 )
367}
368
369/// Re-raise an enriched message as the interpreter does: through
370/// `panic_any`, so the hook prints it once, or prints the short notice
371/// when a downstream reporter renders the full body (SRD-82).
372pub(crate) fn reraise_enriched(enriched: String) -> ! {
373 if PANIC_REPORTING_DOWNSTREAM.load(std::sync::atomic::Ordering::Relaxed) {
374 RERAISE_SHORT.with(|c| c.set(true));
375 }
376 std::panic::panic_any(enriched)
377}
378
379/// The interpreter's enrichment: the node's name and outputs from the
380/// program, the inputs as the `Value`s it was called with.
381fn enrich_eval_panic(
382 payload: Box<dyn std::any::Any + Send>,
383 program: &PolydatProgram,
384 node_idx: usize,
385 inputs: &[Value],
386) -> String {
387 let node_name = program
388 .nodes
389 .get(node_idx)
390 .map(|n| n.meta().name.to_string())
391 .unwrap_or_else(|| format!("<unknown node #{node_idx}>"));
392 let mut output_names: Vec<&str> = program
393 .output_map_iter()
394 .filter_map(|(name, (n_idx, _))| {
395 if *n_idx == node_idx {
396 Some(name.as_str())
397 } else {
398 None
399 }
400 })
401 .collect();
402 output_names.sort();
403 let inputs: Vec<String> = inputs.iter().map(format_value_for_diag).collect();
404 enrich_panic(
405 payload,
406 &node_name,
407 &output_names,
408 program.context(),
409 &inputs,
410 )
411}
412
413/// Format a `Value` into a short diagnostic string. Strings are
414/// quoted + truncated; vectors print their length not contents.
415pub(crate) fn format_value_for_diag(v: &Value) -> String {
416 match v {
417 Value::U64(n) => format!("U64({n})"),
418 Value::F64(n) => format!("F64({n})"),
419 Value::Bool(b) => format!("Bool({b})"),
420 Value::Str(s) => {
421 let trimmed: String = s.chars().take(40).collect();
422 if s.chars().count() > 40 {
423 format!("Str({trimmed:?}…)")
424 } else {
425 format!("Str({trimmed:?})")
426 }
427 }
428 Value::None => "None".to_string(),
429 other => format!("{:?}", other.port_type()),
430 }
431}
432
433/// Shared evaluation state for all Polydat engines. Contains the node
434/// output buffers, input values, and the eval loop.
435/// Engine types wrap this and provide their own invalidation strategy.
436pub struct EngineCore {
437 /// Per-node output value buffers, reused across evaluations.
438 pub(crate) buffers: Vec<Vec<Value>>,
439 /// Per-node: true = cached output is valid, false = needs eval.
440 pub(crate) node_clean: Vec<bool>,
441 /// Current input values (coordinates + captures, all unified).
442 /// For a cell-bound slot this entry is unused: the cell is the
443 /// slot's only register (`read_input` reads it, `set_input`
444 /// publishes to it).
445 pub(crate) inputs: Vec<Value>,
446 /// Default values for each input (used by reset_inputs).
447 pub(crate) input_defaults: Vec<Value>,
448 /// Optional cross-kernel shared cell per input slot. `None`
449 /// = local-only input (the common case). `Some(cell)` =
450 /// the slot is bound to a shared cell; writes propagate
451 /// through the cell to whatever other kernels share it.
452 pub(crate) shared_cells: Vec<Option<SharedCell>>,
453 /// SRD-13f Push B.2 — per-output broadcast cell. Indexed
454 /// by output position in `program.output_list`. `Some(cell)`
455 /// = the output broadcasts its value to descendants via
456 /// the cell whenever the owner pulls the output; `None` =
457 /// no broadcast subscribers were set up (no descendant
458 /// scope binds against this output's name).
459 ///
460 /// `materialize_wiring_from_outer` plumbs the same `Arc<SharedCell>`
461 /// onto the matching input slot on the inner kernel — at
462 /// that point both ends share the storage. Inner reads
463 /// transparently through the cell on every `read_input`;
464 /// outer's `pull` writes the freshly computed value into
465 /// the cell so subsequent inner reads return the current
466 /// value with no traversal.
467 pub(crate) output_cells: Vec<Option<SharedCell>>,
468 /// Pre-allocated scratch buffer for node input gathering.
469 pub(crate) input_scratch: Vec<Value>,
470 /// Per node, the scratch entries the node declared through
471 /// `scratch_layout` (a native cone's own slot buffer): storage
472 /// belongs to the state, never to the node, which is shared by
473 /// every state of the program (axiom S3).
474 pub(crate) node_scratch: Vec<Vec<crate::ast::ScratchBuf>>,
475 /// This scope's intent-dirty bit-vector. One `AtomicU64`
476 /// word per 64 cells allocated by this scope; new words
477 /// are appended on demand by [`Self::allocate_cell_bit`].
478 /// Each cell carries a clone of the specific `Arc<AtomicU64>`
479 /// for its word (and its bit-within-word). Consumer fibers'
480 /// bulk-mask check (per `cross_fiber_invalidation.md` §5)
481 /// groups cells by `Arc::ptr_eq` of their word and ANDs
482 /// the loaded word against the cone's interest mask for
483 /// that word.
484 ///
485 /// The `Vec<Arc<...>>` shape — rather than a single
486 /// `Arc<Vec<AtomicU64>>` — lets cells take a stable
487 /// per-word handle that the scope can grow without
488 /// invalidating any existing cell's reference.
489 pub(crate) scope_intent_words: Vec<Arc<AtomicU64>>,
490 /// Next bit position to allocate from
491 /// [`Self::scope_intent_words`]. Word index is
492 /// `next_cell_bit / 64`; bit within word is
493 /// `next_cell_bit % 64`. Monotonic; bits are never reused
494 /// within a scope's lifetime.
495 pub(crate) next_cell_bit: u32,
496 /// Per-fiber cache of the last revision this engine observed
497 /// for each cell it has read. Keyed by `Arc::as_ptr` of the
498 /// `SharedCellInner`. Sparse; entries are inserted lazily
499 /// on first observation via `check_cell_clean`.
500 ///
501 /// Per-fiber state — no contention. Pointer keys are stable
502 /// for the cell's lifetime; orphaned entries for dropped
503 /// cells are harmless (the handle is never observed again).
504 pub(crate) last_seen: std::collections::HashMap<*const SharedCellInner, u64>,
505 /// Per-node cone metadata for cell-bound input deps. Lazy:
506 /// `None` until first `check_cell_clean` for that node;
507 /// then built once and reused. Cleared in bulk on any
508 /// attach/detach of shared cells.
509 pub(crate) cell_cones: Vec<Option<CellCone>>,
510}
511
512// `last_seen` keys are `*const SharedCellInner` raw pointers,
513// which Rust treats as non-Send/non-Sync. The pointers are
514// only compared by identity (never dereferenced) and each
515// `EngineCore` is owned by exactly one fiber, so the
516// non-thread-safe pointer keys are sound. The Send/Sync
517// markers here cover that gap explicitly.
518unsafe impl Send for EngineCore {}
519unsafe impl Sync for EngineCore {}
520
521impl EngineCore {
522 /// Allocate the next bit position from this scope's
523 /// intent-dirty vector for a newly-created cell. Returns
524 /// the specific word's `Arc<AtomicU64>` plus the bit
525 /// position within that word. Grows
526 /// [`Self::scope_intent_words`] on demand — each new word
527 /// is a freshly-allocated `Arc<AtomicU64>` so existing
528 /// cells' references stay stable.
529 pub(crate) fn allocate_cell_bit(&mut self) -> (Arc<AtomicU64>, u8) {
530 let bit = self.next_cell_bit;
531 let word_idx = (bit / 64) as usize;
532 let bit_in_word = (bit % 64) as u8;
533 while self.scope_intent_words.len() <= word_idx {
534 self.scope_intent_words.push(Arc::new(AtomicU64::new(0)));
535 }
536 let word = self.scope_intent_words[word_idx].clone();
537 self.next_cell_bit += 1;
538 (word, bit_in_word)
539 }
540
541 /// Construct a new `SharedCell` bound to this scope's
542 /// intent-dirty vector. Convenience wrapper that allocates
543 /// a fresh bit and builds the cell — every cell creation
544 /// site goes through here so the scope's bit allocator
545 /// stays the single source of truth.
546 pub(crate) fn make_shared_cell(&mut self, initial: Value) -> SharedCell {
547 let (word, bit) = self.allocate_cell_bit();
548 Arc::new(SharedCellInner::new(initial, word, bit))
549 }
550}
551
552impl EngineCore {
553 /// Read an input slot's current value, transparent to whether
554 /// it's a plain slot or backed by a `SharedCell`. The
555 /// canonical read path used by both `eval_node` and
556 /// `PolydatState::get_input` — there's no separate "refresh" step
557 /// the caller must remember; the cell is queried on every
558 /// read.
559 ///
560 /// Cost: one Mutex lock per read on shared slots; a clone of
561 /// `inputs[idx]` on plain slots (Value's clone is cheap —
562 /// Arc-based for vectors, primitive copy otherwise).
563 #[inline]
564 pub(crate) fn read_input(&self, idx: usize) -> Value {
565 if let Some(cell) = self.shared_cells.get(idx).and_then(|c| c.as_ref()) {
566 return cell.value.lock().unwrap().clone();
567 }
568 self.inputs[idx].clone()
569 }
570
571 /// Build the cone metadata for `node_idx` — the per-scope
572 /// groups of cell-bound input dependencies, derived from
573 /// `program.input_provenance[node_idx]` and the cells
574 /// currently attached on this engine.
575 ///
576 /// Returns an empty `CellCone { groups: [] }` for nodes
577 /// with no cell-bound deps (the common case).
578 fn build_cell_cone(&self, program: &PolydatProgram, node_idx: usize) -> CellCone {
579 let empty = crate::kernel::ProvMask::empty();
580 let prov = program.input_provenance.get(node_idx).unwrap_or(&empty);
581 let mut groups: Vec<CellConeGroup> = Vec::new();
582 // Iterate set bits of `prov` directly: each bit is an
583 // input slot that flows into this node transitively.
584 for input_idx in prov.iter_ones() {
585 let Some(Some(cell)) = self.shared_cells.get(input_idx) else {
586 continue;
587 };
588 // Group by Arc-pointer identity of scope_intent_dirty.
589 let group_idx = groups
590 .iter()
591 .position(|g| Arc::ptr_eq(&g.intent_dirty, &cell.scope_intent_dirty));
592 let i = match group_idx {
593 Some(i) => i,
594 None => {
595 groups.push(CellConeGroup {
596 intent_dirty: cell.scope_intent_dirty.clone(),
597 interest_mask: 0,
598 cells: Vec::new(),
599 });
600 groups.len() - 1
601 }
602 };
603 groups[i].interest_mask |= 1u64 << cell.bit;
604 groups[i].cells.push(CellConeEntry {
605 bit: cell.bit,
606 input_slot: input_idx,
607 });
608 }
609 CellCone { groups }
610 }
611
612 /// Cross-fiber check: return `true` if this fiber's
613 /// `last_seen` is up-to-date for every cell in `node_idx`'s
614 /// cone (no cross-fiber writes since last observation).
615 /// Returns `false` if any cell's revision has advanced,
616 /// updating `last_seen` to reflect the new revisions in
617 /// preparation for the caller's re-evaluation.
618 ///
619 /// Per cross_fiber_invalidation.md §5: bulk-mask check
620 /// (one Acquire load + AND per scope group) early-outs
621 /// when nothing in the scope is dirty; per-cell drill-down
622 /// runs only on set bits.
623 fn check_cell_clean(&mut self, program: &PolydatProgram, node_idx: usize) -> bool {
624 // Lazy build the cone metadata.
625 if self.cell_cones.len() <= node_idx {
626 self.cell_cones.resize_with(node_idx + 1, || None);
627 }
628 if self.cell_cones[node_idx].is_none() {
629 let cone = self.build_cell_cone(program, node_idx);
630 self.cell_cones[node_idx] = Some(cone);
631 }
632
633 // First pass: walk the cone, collect mismatches. The
634 // immutable borrow of `self.cell_cones`,
635 // `self.shared_cells`, and `self.last_seen` coexist
636 // because they're disjoint fields of `self`.
637 let mut dirty: Vec<(*const SharedCellInner, u64, usize)> = Vec::new();
638 {
639 let cone = self.cell_cones[node_idx].as_ref().unwrap();
640 for group in &cone.groups {
641 let intent = group.intent_dirty.load(Ordering::Acquire);
642 let masked = intent & group.interest_mask;
643 if masked == 0 {
644 continue;
645 }
646 for entry in &group.cells {
647 if masked & (1u64 << entry.bit) == 0 {
648 continue;
649 }
650 let Some(Some(cell)) = self.shared_cells.get(entry.input_slot) else {
651 continue;
652 };
653 let r = cell.revision.load(Ordering::Acquire);
654 let ptr = Arc::as_ptr(cell);
655 let prev = self.last_seen.get(&ptr).copied().unwrap_or(0);
656 if r != prev {
657 dirty.push((ptr, r, entry.input_slot));
658 }
659 }
660 }
661 }
662 let clean = dirty.is_empty();
663 // Second pass: update last_seen for every cell whose
664 // revision we observed has advanced. Done in a
665 // separate pass to release the cone borrow above.
666 //
667 // Updating `last_seen` CONSUMES the dirty signal for this
668 // fiber, so the re-evaluation it triggers must reach every
669 // memoized node between the dirty slot and any consumer —
670 // not just the node that happened to check first. The
671 // caller only re-evaluates the CHECKED node; its recursive
672 // upstream walk re-checks each parent's own cone, which now
673 // reads the just-updated `last_seen` and comes back clean,
674 // leaving the intermediate buffers stale — the checked node
675 // then recomputes from stale parents (observed as a
676 // phase-poll predicate memoized at its pre-write value
677 // forever). Mirror `set_input`'s write-side rule on the
678 // read side: a detected cross-fiber write invalidates every
679 // node whose transitive input provenance covers the dirty
680 // slot.
681 if !clean {
682 // Exact multi-word mask: slots >= 64 invalidate too
683 // (the one-word form silently SKIPPED them — a latent
684 // under-invalidation on >64-input scopes).
685 let mut dirty_mask = crate::kernel::ProvMask::empty();
686 for (ptr, r, slot) in dirty {
687 self.last_seen.insert(ptr, r);
688 dirty_mask.set(slot);
689 }
690 for node_idx in 0..program.nodes.len() {
691 if program
692 .input_provenance
693 .get(node_idx)
694 .is_some_and(|prov| prov.intersects(&dirty_mask))
695 {
696 self.node_clean[node_idx] = false;
697 }
698 }
699 }
700 clean
701 }
702
703 /// Mark `cell_cones` as stale. Called after any change to
704 /// `shared_cells` that could affect the per-node cone
705 /// metadata (attach, detach). Next `check_cell_clean` on
706 /// any node will rebuild on demand.
707 pub(crate) fn invalidate_cell_cones(&mut self) {
708 for cone in self.cell_cones.iter_mut() {
709 *cone = None;
710 }
711 }
712
713 /// Evaluate a node by index. Shared by all engines.
714 /// Checks the clean flag, recursively evaluates upstream, gathers
715 /// inputs, calls node.eval(), marks clean.
716 pub fn eval_node(&mut self, program: &PolydatProgram, node_idx: usize) {
717 if self.node_clean[node_idx] {
718 // Memoization hit candidate — confirm cell-bound
719 // inputs in this node's cone are still at the
720 // revisions this fiber last observed. If any
721 // producer fiber has bumped a cell's revision since
722 // then, force a re-eval (the cache is stale even
723 // though `node_clean` is true) per
724 // cross_fiber_invalidation.md §5.
725 if self.check_cell_clean(program, node_idx) {
726 return;
727 }
728 self.node_clean[node_idx] = false;
729 }
730
731 let wiring = &program.wiring[node_idx];
732 for source in wiring.iter() {
733 if let WireSource::NodeOutput(upstream_idx, _) = source {
734 self.eval_node(program, *upstream_idx);
735 }
736 }
737
738 for (i, source) in wiring.iter().enumerate() {
739 self.input_scratch[i] = match source {
740 // `read_input` transparently reads the cell for
741 // `shared`-bound slots, so per-cycle eval picks
742 // up cross-kernel writes without any explicit
743 // refresh.
744 WireSource::Input(idx) => self.read_input(*idx),
745 WireSource::NodeOutput(upstream_idx, port_idx) => {
746 self.buffers[*upstream_idx][*port_idx].clone()
747 }
748 };
749 }
750
751 let input_count = wiring.len();
752
753 // SRD-74 Rule 1 — None propagation lifted to the kernel
754 // level. Any node whose inputs include `Value::None`
755 // emits `Value::None` on every output without invoking
756 // the node's `eval`. This holds the SQL-NULL / Rust
757 // `Option::?` propagation rule uniformly for ALL GK
758 // nodes, avoiding the dozens of duplicate per-node
759 // `if matches!(input, Value::None)` checks. Individual
760 // nodes (e.g. `Printf`) keep their checks redundant but
761 // harmless — the kernel guard fires first.
762 //
763 // Opt-out: nodes whose semantics explicitly consume
764 // `Value::None` (coalesce-style `default_or`, explicit
765 // optionality handlers per SRD-74 Rule 2) override
766 // `PolydatNode::accepts_none_inputs` to skip this guard. Such
767 // nodes handle `None` in their own `eval`.
768 let node_ref = &*program.nodes[node_idx];
769 if !node_ref.accepts_none_inputs()
770 && self.input_scratch[..input_count]
771 .iter()
772 .any(|v| matches!(v, Value::None))
773 {
774 for slot in &mut self.buffers[node_idx] {
775 *slot = Value::None;
776 }
777 self.node_clean[node_idx] = true;
778 return;
779 }
780
781 // Wrap the node's eval in catch_unwind so a node-level
782 // panic (e.g. `Value::as_u64` on a Str) can be re-raised
783 // with the diagnostic context the user actually needs:
784 // which node panicked, which output(s) it feeds, what
785 // the input values were, and where in the source the
786 // node came from. Without this, the fiber-level catcher
787 // sees only the bare message — "expected U64, got Str"
788 // — and the user has no way to find the offending
789 // binding short of bisecting the workload.
790 //
791 // Cost: one catch_unwind frame per slow-path node eval.
792 // The JIT path doesn't go through here. On the success
793 // path the frame is a few stack words; on the panic
794 // path it's strictly an improvement over what the
795 // user sees today.
796 //
797 // The capture guard suppresses the std panic hook for
798 // the duration: without it, the hook prints the BARE
799 // payload ("expected U64, got F64" + backtrace) at the
800 // original panic site, before enrichment exists, and
801 // that raw print is the loudest thing the user sees.
802 // Re-raising with `panic_any` (not `resume_unwind`)
803 // fires the hook again — now unsuppressed — so the one
804 // message that prints is the enriched one.
805 let guard = EvalPanicCaptureGuard::arm();
806 let payload = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
807 program.nodes[node_idx].eval_in(
808 &mut self.node_scratch[node_idx],
809 &self.input_scratch[..input_count],
810 &mut self.buffers[node_idx],
811 );
812 }));
813 drop(guard);
814 if let Err(e) = payload {
815 // A native cone re-raises its member's failure already
816 // enriched with the member's name, its inputs, and this
817 // program's context (A7); the cone itself is not a frame,
818 // so the report reads as it does on every other engine.
819 if program.nodes[node_idx].fusion_subgraph().is_some()
820 && e.downcast_ref::<String>()
821 .is_some_and(|s| s.contains("↳ in node"))
822 {
823 let enriched = *e.downcast::<String>().expect("checked above");
824 reraise_enriched(enriched);
825 }
826 let enriched =
827 enrich_eval_panic(e, program, node_idx, &self.input_scratch[..input_count]);
828 reraise_enriched(enriched);
829 }
830 self.node_clean[node_idx] = true;
831 }
832
833 /// Pull a named output.
834 pub fn pull(&mut self, program: &PolydatProgram, output_name: &str) -> &Value {
835 let (node_idx, port_idx) = *program
836 .output_map
837 .get(output_name)
838 .unwrap_or_else(|| panic!("unknown output variate: {output_name}"));
839 self.eval_node(program, node_idx);
840 // SRD-13f Push B.2: broadcast the freshly computed
841 // value through this output's cell (if attached) so
842 // descendant kernels that bound their matching input
843 // slot to the same cell see the current value on
844 // their next read.
845 if let Some(output_idx) = program.output_index(output_name)
846 && let Some(Some(cell)) = self.output_cells.get(output_idx)
847 {
848 let v = self.buffers[node_idx][port_idx].clone();
849 // `publish` does the mutex write + revision bump +
850 // intent-bit set in three Release stores so the
851 // descendant's cone walker observes the change on
852 // its next read (cross_fiber_invalidation.md §5).
853 cell.publish(v);
854 }
855 &self.buffers[node_idx][port_idx]
856 }
857
858 /// SRD-13f Push B.2 — allocate broadcast cells for every
859 /// output in `program`. Idempotent: if cells are already
860 /// allocated (size matches the program's output count),
861 /// the call is a no-op. Initial cell value is taken from
862 /// the current buffer (typically `Value::None` at
863 /// construction, before any pull has fired).
864 ///
865 /// Called from kernel constructors and from
866 /// `materialize_wiring_from_outer`-style operations that materialize
867 /// new descendants — the inner side needs the cell to
868 /// exist before it can attach to its input slot.
869 pub(crate) fn seed_output_cells(&mut self, program: &PolydatProgram) {
870 let n = program.output_names().len();
871 if self.output_cells.len() == n {
872 return;
873 }
874 // Two-pass to avoid borrowing `self` immutably (for
875 // buffer lookups) while also borrowing it mutably (for
876 // `make_shared_cell`). First collect initial values,
877 // then construct the cells.
878 let initials: Vec<Value> = (0..n)
879 .map(|i| {
880 let name = &program.output_list()[i].0;
881 let (node_idx, port_idx) = program.output_map[name];
882 // Defensive bounds-check: some construction paths
883 // (raw state, partial programs) may not populate
884 // buffers for every node referenced in the output
885 // map. Seed with `Value::None` rather than panic.
886 self.buffers
887 .get(node_idx)
888 .and_then(|b| b.get(port_idx))
889 .cloned()
890 .unwrap_or(Value::None)
891 })
892 .collect();
893 self.output_cells = initials
894 .into_iter()
895 .map(|init| Some(self.make_shared_cell(init)))
896 .collect();
897 }
898
899 /// Output broadcast cell for the named output, if seeded.
900 pub(crate) fn output_cell(&self, program: &PolydatProgram, name: &str) -> Option<SharedCell> {
901 let idx = program.output_index(name)?;
902 self.output_cells.get(idx).and_then(|c| c.clone())
903 }
904}
905
906// =================================================================
907// PolydatState: dependent-list engine (default, O(affected) invalidation)
908// =================================================================
909
910/// Polydat evaluation engine using precomputed per-input dependent lists.
911///
912/// On `set_input()`, only nodes that depend on the changed input
913/// are dirtied. O(affected_nodes) per input change.
914/// This is the interpreter's state, the default of its three; the
915/// default engine for engine-less entry points is the compiled P3 engine.
916pub struct PolydatState {
917 /// Shared evaluation core (buffers, clean flags, inputs).
918 pub core: EngineCore,
919 /// Per-input dependent node lists for O(affected) invalidation.
920 input_dependents: Vec<Vec<usize>>,
921 /// Indices of non-deterministic nodes (zero-provenance, no declared inputs).
922 ///
923 /// These nodes produce a different value on every evaluation (e.g.,
924 /// `counter()`, `current_epoch_millis()`). They are unconditionally
925 /// marked dirty on every `set_input()` call so they are never cached.
926 nondeterministic_nodes: Vec<usize>,
927}
928
929impl PolydatState {
930 /// Construct a PolydatState from its component parts.
931 pub(crate) fn from_parts(
932 core: EngineCore,
933 input_dependents: Vec<Vec<usize>>,
934 nondeterministic_nodes: Vec<usize>,
935 ) -> Self {
936 Self {
937 core,
938 input_dependents,
939 nondeterministic_nodes,
940 }
941 }
942
943 /// Set all coordinate inputs at once. Wraps each u64 as
944 /// `Value::U64` and sets them at indices 0..N with per-input
945 /// change detection.
946 pub fn set_inputs(&mut self, coords: &[u64]) {
947 self.write_coordinates(coords);
948 }
949
950 /// Write the coordinates: what construction does to seed a state's
951 /// folded constants. A host's write is [`Self::set_inputs`].
952 pub(crate) fn seed_inputs(&mut self, coords: &[u64]) {
953 self.write_coordinates(coords);
954 }
955
956 fn write_coordinates(&mut self, coords: &[u64]) {
957 for (i, &c) in coords.iter().enumerate().take(self.core.inputs.len()) {
958 self.core.inputs[i] = Value::U64(c);
959 // Unconditional invalidation: the write itself is the
960 // signal — see `set_input` for the rationale.
961 if i < self.input_dependents.len() {
962 for &node_idx in &self.input_dependents[i] {
963 self.core.node_clean[node_idx] = false;
964 }
965 }
966 }
967 // Non-deterministic nodes must always re-evaluate.
968 for &idx in &self.nondeterministic_nodes {
969 self.core.node_clean[idx] = false;
970 }
971 }
972
973 /// Set a single input by index, dirtying only dependent nodes.
974 ///
975 /// Single-register semantics: a cell-bound slot's only
976 /// register IS the cell — `set_input` writes through the
977 /// cell. A non-cell slot's register is the local
978 /// `inputs[idx]` array. There's no second snapshot kept in
979 /// lockstep with the cell; reads always go to whichever is
980 /// the slot's register.
981 ///
982 /// Dependents-marking is the dependent-list invalidation
983 /// strategy carried by `PolydatState`; it's the write-side
984 /// half of the engine's dirty-tracking. Other engines
985 /// (`RawState`, `ProvScanState`) implement different
986 /// strategies — see their own `set_inputs` impls.
987 pub fn set_input(&mut self, idx: usize, value: Value) {
988 if let Some(cell) = self.core.shared_cells.get(idx).and_then(|c| c.as_ref()) {
989 // Cell-bound slot: the cell is the register. We do
990 // NOT mirror the value into `inputs[idx]`; that
991 // array slot is unused for cell-bound inputs.
992 //
993 // `publish` does the mutex write + revision bump +
994 // intent-bit set in three Release stores so the
995 // any other fiber's cone walker observes the
996 // change on its next read
997 // (cross_fiber_invalidation.md §5).
998 cell.publish(value);
999 } else {
1000 self.core.inputs[idx] = value;
1001 }
1002 // Mark every transitive dependent dirty unconditionally.
1003 // The act of writing an input IS the invalidation
1004 // signal — we don't gate on value equality because (a)
1005 // structural equality on rich Value variants
1006 // (Json/Bytes/VecF32) is expensive enough to defeat
1007 // the purpose of the optimisation, and (b) a same-
1008 // value rewrite is still a legitimate "the upstream
1009 // owner asked for a re-evaluation" signal that
1010 // downstream side-effecting nodes (`log_*`, audit
1011 // emitters, time-stamped observers) MUST honour.
1012 let dirty_debug = nbrs_dirty_debug_enabled();
1013 if idx < self.input_dependents.len() {
1014 if dirty_debug {
1015 eprintln!(
1016 "DIRTY: set_input idx={idx} input_count={} dependents_for_idx={} \
1017 total_input_dependents_len={}",
1018 self.core.inputs.len(),
1019 self.input_dependents[idx].len(),
1020 self.input_dependents.len()
1021 );
1022 }
1023 for &node_idx in &self.input_dependents[idx] {
1024 self.core.node_clean[node_idx] = false;
1025 }
1026 } else if dirty_debug {
1027 eprintln!(
1028 "DIRTY: set_input idx={idx} OUT_OF_RANGE input_dependents_len={}",
1029 self.input_dependents.len()
1030 );
1031 }
1032 // Non-deterministic nodes must always re-evaluate.
1033 for &idx in &self.nondeterministic_nodes {
1034 self.core.node_clean[idx] = false;
1035 }
1036 }
1037
1038 /// Read the value of an input by index.
1039 ///
1040 /// Single-register read: cell-bound slots return the cell's
1041 /// current value; non-cell slots return the local register.
1042 /// One canonical value per slot, no stale snapshot.
1043 pub fn get_input(&self, idx: usize) -> Value {
1044 self.core.read_input(idx)
1045 }
1046
1047 /// Alias for [`Self::get_input`]; kept for legacy callers
1048 /// that picked the more explicit name. Both read the cell
1049 /// when one is attached.
1050 pub fn read_input_value(&self, idx: usize) -> Value {
1051 self.core.read_input(idx)
1052 }
1053
1054 /// Attach a `SharedCell` to an input slot.
1055 ///
1056 /// After this call the cell becomes the slot's sole
1057 /// register: reads via `read_input` go through the cell,
1058 /// `set_input` writes through the cell. The local
1059 /// `inputs[idx]` array entry for this slot is unused for
1060 /// cell-bound slots — there is no second register kept in
1061 /// lockstep.
1062 ///
1063 /// Dependents are dirtied because the slot's effective
1064 /// value just changed from the local default to whatever
1065 /// the cell currently holds.
1066 pub fn attach_shared_cell(&mut self, idx: usize, cell: SharedCell) {
1067 if idx >= self.core.shared_cells.len() {
1068 self.core.shared_cells.resize(idx + 1, None);
1069 }
1070 self.core.shared_cells[idx] = Some(cell);
1071 if idx < self.input_dependents.len() {
1072 for &node_idx in &self.input_dependents[idx] {
1073 self.core.node_clean[node_idx] = false;
1074 }
1075 }
1076 // Cone metadata depends on which slots have cells; the
1077 // new attachment invalidates any cached cone groups.
1078 // Next `check_cell_clean` per node rebuilds on demand
1079 // per cross_fiber_invalidation.md §3.1.
1080 self.core.invalidate_cell_cones();
1081 }
1082
1083 /// Returns the `SharedCell` attached to an input slot, if any.
1084 /// Used by `materialize_wiring_from_outer` to share an existing cell with
1085 /// inner kernels.
1086 pub fn shared_cell(&self, idx: usize) -> Option<SharedCell> {
1087 self.core.shared_cells.get(idx).and_then(|c| c.clone())
1088 }
1089
1090 /// Reset a range of inputs to their defaults. Used at stanza
1091 /// boundaries to prevent capture leakage across stanzas.
1092 /// `from_idx` is typically `coord_count` (skip coordinates,
1093 /// reset only capture inputs).
1094 ///
1095 /// Cell-bound slots are skipped: the cell is cross-kernel
1096 /// shared state with its own lifecycle (managed by the
1097 /// owning ancestor scope), and a stanza-local reset must
1098 /// not clobber other kernels' views.
1099 pub fn reset_inputs_from(&mut self, from_idx: usize) {
1100 for i in from_idx..self.core.inputs.len() {
1101 // Cell-bound slots: the cell is the register, owned
1102 // by the ancestor that declared `shared X := init`.
1103 // Don't touch.
1104 if self.core.shared_cells.get(i).is_some_and(|c| c.is_some()) {
1105 continue;
1106 }
1107 if self.core.inputs[i] != self.core.input_defaults[i] {
1108 self.core.inputs[i] = self.core.input_defaults[i].clone();
1109 if i < self.input_dependents.len() {
1110 for &node_idx in &self.input_dependents[i] {
1111 self.core.node_clean[node_idx] = false;
1112 }
1113 }
1114 }
1115 }
1116 }
1117
1118 /// Mark every node dirty and leave the inputs as they are: every
1119 /// node reruns at the next pull, as if the cycle had moved. What
1120 /// `Kernel::invalidate_all` means on every engine; a host that
1121 /// wants the inputs back at their defaults calls
1122 /// [`Self::reset_inputs_from`] as well.
1123 pub fn invalidate_all(&mut self) {
1124 self.core.node_clean.fill(false);
1125 }
1126
1127 /// Pull a named output variate from the program.
1128 pub fn pull(&mut self, program: &PolydatProgram, output_name: &str) -> &Value {
1129 self.core.pull(program, output_name)
1130 }
1131
1132 /// Pre-populate a node's output buffer slot and mark it clean,
1133 /// suppressing on-demand evaluation. Used by the scope-init
1134 /// pass (SRD 11 §"Init Binding Contract" Plan B) to seed
1135 /// per-fiber states with init binding values that the
1136 /// activation kernel already evaluated, so each fiber doesn't
1137 /// re-fire the eval at first pull.
1138 pub fn seed_node_buffer(&mut self, node_idx: usize, port_idx: usize, value: Value) {
1139 if node_idx >= self.core.buffers.len() {
1140 return;
1141 }
1142 if port_idx >= self.core.buffers[node_idx].len() {
1143 return;
1144 }
1145 self.core.buffers[node_idx][port_idx] = value;
1146 self.core.node_clean[node_idx] = true;
1147 }
1148
1149 /// Read a node's output buffer slot. Used by the scope-init
1150 /// pass to extract a pre-pulled init binding value from one
1151 /// state and seed it into another.
1152 pub fn node_buffer(&self, node_idx: usize, port_idx: usize) -> Option<&Value> {
1153 self.core
1154 .buffers
1155 .get(node_idx)
1156 .and_then(|ports| ports.get(port_idx))
1157 }
1158
1159 /// Pull an output by index (declaration order). Only evaluates
1160 /// the computation cone for this specific output.
1161 pub fn pull_by_index(&mut self, program: &PolydatProgram, output_idx: usize) -> &Value {
1162 let (node_idx, port_idx) = program.resolve_output_by_index(output_idx);
1163 self.core.eval_node(program, node_idx);
1164 &self.core.buffers[node_idx][port_idx]
1165 }
1166
1167 /// Pull all outputs in declaration order.
1168 pub fn pull_all<'a>(&'a mut self, program: &PolydatProgram) -> Vec<&'a Value> {
1169 for i in 0..program.output_count() {
1170 let (node_idx, _) = program.resolve_output_by_index(i);
1171 self.core.eval_node(program, node_idx);
1172 }
1173 (0..program.output_count())
1174 .map(|i| {
1175 let (ni, pi) = program.resolve_output_by_index(i);
1176 &self.core.buffers[ni][pi]
1177 })
1178 .collect()
1179 }
1180
1181 /// Create a memoized accessor for a named subset of outputs.
1182 /// Resolves names to indices once; subsequent access uses indices only.
1183 pub fn accessor(program: &PolydatProgram, names: &[&str]) -> OutputAccessor {
1184 let indices: Vec<usize> = names
1185 .iter()
1186 .filter_map(|n| program.output_index(n))
1187 .collect();
1188 OutputAccessor { indices }
1189 }
1190
1191 /// Evaluate a node by index (exposed for constant folding in PolydatProgram).
1192 pub(crate) fn eval_node_public(&mut self, program: &PolydatProgram, node_idx: usize) {
1193 self.core.eval_node(program, node_idx);
1194 }
1195}
1196
1197/// Memoized output accessor for a named subset of outputs.
1198///
1199/// Created once from output names via `PolydatState::accessor()`.
1200/// Subsequent pulls use pre-resolved indices — no name lookups.
1201pub struct OutputAccessor {
1202 indices: Vec<usize>,
1203}
1204
1205impl OutputAccessor {
1206 /// Pull all outputs in this accessor from the given state.
1207 pub fn pull_all<'a>(
1208 &self,
1209 state: &'a mut PolydatState,
1210 program: &PolydatProgram,
1211 ) -> Vec<&'a Value> {
1212 for &idx in &self.indices {
1213 let (node_idx, _) = program.resolve_output_by_index(idx);
1214 state.core.eval_node(program, node_idx);
1215 }
1216 self.indices
1217 .iter()
1218 .map(|&idx| {
1219 let (ni, pi) = program.resolve_output_by_index(idx);
1220 &state.core.buffers[ni][pi]
1221 })
1222 .collect()
1223 }
1224
1225 /// Number of outputs in this accessor.
1226 pub fn len(&self) -> usize {
1227 self.indices.len()
1228 }
1229
1230 /// Whether this accessor has no outputs.
1231 pub fn is_empty(&self) -> bool {
1232 self.indices.is_empty()
1233 }
1234}
1235
1236// =================================================================
1237// RawState: no provenance engine (all nodes dirty every eval)
1238// =================================================================
1239
1240/// Polydat evaluation engine with no provenance. Every `set_inputs()`
1241/// marks all nodes dirty. Baseline for benchmarking provenance overhead.
1242pub struct RawState {
1243 /// Shared evaluation core.
1244 pub core: EngineCore,
1245}
1246
1247impl RawState {
1248 /// Set new input values and mark all nodes dirty (no provenance check).
1249 pub fn set_inputs(&mut self, coords: &[u64]) {
1250 for (i, &c) in coords.iter().enumerate().take(self.core.inputs.len()) {
1251 self.core.inputs[i] = Value::U64(c);
1252 }
1253 self.core.node_clean.fill(false);
1254 }
1255
1256 /// Pull a named output variate from the program.
1257 pub fn pull(&mut self, program: &PolydatProgram, output_name: &str) -> &Value {
1258 self.core.pull(program, output_name)
1259 }
1260}
1261
1262// =================================================================
1263// ProvScanState: provenance-scan engine (O(all) invalidation)
1264// =================================================================
1265
1266/// Polydat evaluation engine using provenance bitmask scanning.
1267///
1268/// On `set_inputs()`, scans ALL nodes and checks each node's
1269/// provenance bitmask against the changed-inputs mask.
1270/// O(all_nodes) per input change regardless of how many changed.
1271pub struct ProvScanState {
1272 /// Shared evaluation core.
1273 pub core: EngineCore,
1274 input_provenance: Vec<crate::kernel::ProvMask>,
1275 /// Indices of non-deterministic nodes.
1276 nondeterministic_nodes: Vec<usize>,
1277}
1278
1279impl ProvScanState {
1280 /// Construct a ProvScanState from its component parts.
1281 pub(crate) fn from_parts(
1282 core: EngineCore,
1283 input_provenance: Vec<crate::kernel::ProvMask>,
1284 nondeterministic_nodes: Vec<usize>,
1285 ) -> Self {
1286 Self {
1287 core,
1288 input_provenance,
1289 nondeterministic_nodes,
1290 }
1291 }
1292
1293 /// Set new input values and invalidate affected nodes.
1294 pub fn set_inputs(&mut self, coords: &[u64]) {
1295 let mut mask = crate::kernel::ProvMask::empty();
1296 for (i, &c) in coords.iter().enumerate().take(self.core.inputs.len()) {
1297 self.core.inputs[i] = Value::U64(c);
1298 // Unconditional: writing the input IS the
1299 // invalidation signal regardless of value equality.
1300 mask.set(i);
1301 }
1302 if !mask.is_zero() {
1303 for (i, clean) in self.core.node_clean.iter_mut().enumerate() {
1304 if *clean && self.input_provenance[i].intersects(&mask) {
1305 *clean = false;
1306 }
1307 }
1308 }
1309 for &idx in &self.nondeterministic_nodes {
1310 self.core.node_clean[idx] = false;
1311 }
1312 }
1313
1314 /// Pull a named output variate from the program.
1315 pub fn pull(&mut self, program: &PolydatProgram, output_name: &str) -> &Value {
1316 self.core.pull(program, output_name)
1317 }
1318}