diff --git a/crates/aura-core/src/any.rs b/crates/aura-core/src/any.rs index adacb7e..310cfba 100644 --- a/crates/aura-core/src/any.rs +++ b/crates/aura-core/src/any.rs @@ -5,6 +5,7 @@ use crate::column::Column; use crate::error::KindMismatch; use crate::scalar::{Scalar, ScalarKind, Timestamp}; +use crate::cell::Cell; /// A type-erased edge over the four scalar columns (C7). pub enum AnyColumn { @@ -81,6 +82,21 @@ impl AnyColumn { } } + /// Branch-free, infallible hot-path push of a bare [`Cell`] into this column's + /// concrete `Column`, reinterpreting the cell by *this* column's kind. The + /// inter-node forward uses this: the edge's `from_field`→slot kind match is + /// verified once at bootstrap (C7), so no per-value kind check is needed here + /// — unlike [`push`](Self::push), which keeps the kind guard for the + /// ingestion boundary. + pub fn push_cell(&mut self, c: Cell) { + match self { + AnyColumn::I64(col) => col.push(c.i64()), + AnyColumn::F64(col) => col.push(c.f64()), + AnyColumn::Bool(col) => col.push(c.bool()), + AnyColumn::Ts(col) => col.push(c.ts()), + } + } + /// Read one type-erased value (0 = newest). `None` if cold / out of range. pub fn get(&self, k: usize) -> Option { match self { @@ -170,6 +186,28 @@ mod tests { assert_eq!(e.get(2), None); } + #[test] + fn push_cell_round_trips_by_column_kind() { + // push_cell reinterprets the bare cell by the column's own kind; reading + // back through `get` reconstructs the self-describing Scalar. + let mut f = AnyColumn::with_capacity(ScalarKind::F64, 2); + f.push_cell(Cell::from_f64(1.5)); + assert_eq!(f.get(0), Some(Scalar::f64(1.5))); + assert_eq!(f.run_count(), 1); + + let mut i = AnyColumn::with_capacity(ScalarKind::I64, 2); + i.push_cell(Cell::from_i64(42)); + assert_eq!(i.get(0), Some(Scalar::i64(42))); + + let mut b = AnyColumn::with_capacity(ScalarKind::Bool, 2); + b.push_cell(Cell::from_bool(true)); + assert_eq!(b.get(0), Some(Scalar::bool(true))); + + let mut t = AnyColumn::with_capacity(ScalarKind::Timestamp, 2); + t.push_cell(Cell::from_ts(Timestamp(7))); + assert_eq!(t.get(0), Some(Scalar::ts(Timestamp(7)))); + } + #[test] fn wrong_kind_push_is_rejected() { let mut edge = AnyColumn::with_capacity(ScalarKind::I64, 8); diff --git a/crates/aura-core/src/node.rs b/crates/aura-core/src/node.rs index 15e02f9..c071a43 100644 --- a/crates/aura-core/src/node.rs +++ b/crates/aura-core/src/node.rs @@ -10,7 +10,7 @@ //! into the sweep's flat, path-qualified param-space (C8/C23). Identity is //! positional (slot); the name is a non-load-bearing debug symbol. -use crate::{Ctx, Scalar, ScalarKind}; +use crate::{Cell, Ctx, Scalar, ScalarKind}; /// The firing policy of one input (C6): how the engine decides whether a node /// re-evaluates this cycle. A node fires when *any* of its input groups fires @@ -261,7 +261,7 @@ pub trait Node { /// declared `signature().inputs.len()`. fn lookbacks(&self) -> Vec; - fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]>; + fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]>; /// A one-line, **non-load-bearing** render label (C23): a debug symbol for /// tracing / graph rendering (#13), never read by the run loop and never /// part of wiring (which is by index). Overrides SHOULD carry the node's @@ -285,7 +285,7 @@ mod tests { fn lookbacks(&self) -> Vec { vec![] } - fn eval(&mut self, _ctx: Ctx<'_>) -> Option<&[Scalar]> { + fn eval(&mut self, _ctx: Ctx<'_>) -> Option<&[Cell]> { None } } @@ -407,7 +407,7 @@ mod tests { fn lookbacks(&self) -> Vec { vec![] } - fn eval(&mut self, _ctx: Ctx<'_>) -> Option<&[Scalar]> { + fn eval(&mut self, _ctx: Ctx<'_>) -> Option<&[Cell]> { None } fn label(&self) -> String { diff --git a/crates/aura-engine/src/blueprint.rs b/crates/aura-engine/src/blueprint.rs index b646752..733866f 100644 --- a/crates/aura-engine/src/blueprint.rs +++ b/crates/aura-engine/src/blueprint.rs @@ -813,7 +813,7 @@ mod tests { use super::*; use crate::test_fixtures::{composite_sma_cross_harness, synthetic_prices}; use crate::VecSource; - use aura_core::{Ctx, FieldSpec, Firing, NodeSchema, Timestamp}; + use aura_core::{Cell, Ctx, FieldSpec, Firing, NodeSchema, Timestamp}; use aura_std::{Ema, Exposure, Recorder, SimBroker, Sma, Sub}; use std::sync::mpsc; @@ -1054,37 +1054,37 @@ mod tests { /// A 2-input f64 node, one f64 output. Test-local fixture (C9: examples for the /// engine's own tests, no speculative `aura-std` surface). struct Join2 { - out: [Scalar; 1], + out: [Cell; 1], } impl Node for Join2 { fn lookbacks(&self) -> Vec { vec![1, 1] } - fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> { + fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> { let a = ctx.f64_in(0); let b = ctx.f64_in(1); if a.is_empty() || b.is_empty() { return None; } - self.out[0] = Scalar::f64(a[0] + b[0]); + self.out[0] = Cell::from_f64(a[0] + b[0]); Some(&self.out) } } /// A 1-input f64 node, one f64 output. Test-local fixture. struct Pass1 { - out: [Scalar; 1], + out: [Cell; 1], } impl Node for Pass1 { fn lookbacks(&self) -> Vec { vec![1] } - fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> { + fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> { let w = ctx.f64_in(0); if w.is_empty() { return None; } - self.out[0] = Scalar::f64(w[0]); + self.out[0] = Cell::from_f64(w[0]); Some(&self.out) } } @@ -1095,7 +1095,7 @@ mod tests { fn lookbacks(&self) -> Vec { vec![1] } - fn eval(&mut self, _ctx: Ctx<'_>) -> Option<&[Scalar]> { + fn eval(&mut self, _ctx: Ctx<'_>) -> Option<&[Cell]> { None } } @@ -1107,7 +1107,7 @@ mod tests { fn lookbacks(&self) -> Vec { vec![1] } - fn eval(&mut self, _ctx: Ctx<'_>) -> Option<&[Scalar]> { + fn eval(&mut self, _ctx: Ctx<'_>) -> Option<&[Cell]> { None } } @@ -1116,7 +1116,7 @@ mod tests { PrimitiveBuilder::new( "Pass1", NodeSchema { inputs: vec![f64_any()], output: out_v(), params: vec![] }, - |_| Box::new(Pass1 { out: [Scalar::f64(0.0)] }), + |_| Box::new(Pass1 { out: [Cell::from_f64(0.0)] }), ) .into() } @@ -1124,7 +1124,7 @@ mod tests { PrimitiveBuilder::new( "Join2", NodeSchema { inputs: vec![f64_any(), f64_any()], output: out_v(), params: vec![] }, - |_| Box::new(Join2 { out: [Scalar::f64(0.0)] }), + |_| Box::new(Join2 { out: [Cell::from_f64(0.0)] }), ) .into() } @@ -2280,7 +2280,7 @@ mod tests { let paramless_sma = PrimitiveBuilder::new( "Pass1", NodeSchema { inputs: vec![f64_any()], output: out_v(), params: vec![] }, - |_| Box::new(Pass1 { out: [Scalar::f64(0.0)] }), + |_| Box::new(Pass1 { out: [Cell::from_f64(0.0)] }), ) .named("sma"); let asym = Composite::new( diff --git a/crates/aura-engine/src/graph_model.rs b/crates/aura-engine/src/graph_model.rs index 8014c20..1d061b6 100644 --- a/crates/aura-engine/src/graph_model.rs +++ b/crates/aura-engine/src/graph_model.rs @@ -272,7 +272,7 @@ mod tests { use super::*; use crate::{Edge, OutField, Role, Target}; use aura_core::{ - FieldSpec, Node, NodeSchema, PortSpec, PrimitiveBuilder, Scalar, + Cell, FieldSpec, Node, NodeSchema, PortSpec, PrimitiveBuilder, Scalar, }; use aura_std::{Exposure, Recorder, Sma, Sub}; use std::sync::mpsc; @@ -283,7 +283,7 @@ mod tests { fn lookbacks(&self) -> Vec { vec![] } - fn eval(&mut self, _c: aura_core::Ctx<'_>) -> Option<&[Scalar]> { + fn eval(&mut self, _c: aura_core::Ctx<'_>) -> Option<&[Cell]> { None } } diff --git a/crates/aura-engine/src/harness.rs b/crates/aura-engine/src/harness.rs index e67125f..1c2d57a 100644 --- a/crates/aura-engine/src/harness.rs +++ b/crates/aura-engine/src/harness.rs @@ -7,7 +7,7 @@ //! mutability, no per-cycle allocation (C1/C7). Each node owns its input columns //! (the cycle-0002 shape), so `Ctx` borrows them read-only and additionally //! carries the cycle timestamp (`ctx.now()`, C4). A node's `eval` returns a -//! borrowed record (`Option<&[Scalar]>`); each out-edge forwards one field of it +//! borrowed record (`Option<&[Cell]>`); each out-edge forwards one field of it //! (`Edge::from_field`) into a consumer slot, so the K fields are co-fresh (C6). //! //! Firing (C5/C6) gates re-evaluation. Two read clocks drive it, both stamped per @@ -21,7 +21,7 @@ //! sources are four cycles, so RustAst's `cycle_id`-equality barrier could never //! fire across sources — the timestamp generalizes it faithfully. -use aura_core::{AnyColumn, Ctx, Firing, Node, NodeSchema, Scalar, ScalarKind, Timestamp}; +use aura_core::{AnyColumn, Cell, Ctx, Firing, Node, NodeSchema, Scalar, ScalarKind, Timestamp}; /// Forwards one field (`from_field`) of a producer's output record into a /// consumer's input slot. Consuming a whole record is N such edges, one per field @@ -370,7 +370,7 @@ impl Harness { let Harness { nodes, topo, out_edges, sources } = self; let mut cycle_id: u64 = 0; - let mut scratch: Vec = Vec::new(); + let mut scratch: Vec = Vec::new(); loop { // pick the live source head with the smallest (timestamp, source index). @@ -413,7 +413,7 @@ impl Harness { if !fired { continue; // hold: no eval, no push } - let result: Option<&[Scalar]> = { + let result: Option<&[Cell]> = { let nb = &mut nodes[nidx]; nb.node.eval(Ctx::new(&nb.inputs, ts)) }; @@ -423,9 +423,7 @@ impl Harness { scratch.extend_from_slice(row); for e in out_edges[nidx].iter() { let nb = &mut nodes[e.to]; - nb.inputs[e.slot] - .push(scratch[e.from_field]) - .expect("edge kind checked at wiring"); + nb.inputs[e.slot].push_cell(scratch[e.from_field]); nb.slots[e.slot] = SlotState { fresh_at: cycle_id, last_ts: ts }; } } @@ -618,7 +616,7 @@ mod tests { /// Mode A as-of join: a 2-input f64 sum that fires whenever either input is /// fresh, holding the other. Warm-up returns None until both have a value. struct AsOfSum { - out: [Scalar; 1], + out: [Cell; 1], } impl AsOfSum { fn sig() -> NodeSchema { @@ -633,13 +631,13 @@ mod tests { fn lookbacks(&self) -> Vec { vec![1, 1] } - fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> { + fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> { let a = ctx.f64_in(0); let b = ctx.f64_in(1); if a.is_empty() || b.is_empty() { return None; } - self.out[0] = Scalar::f64(a[0] + b[0]); + self.out[0] = Cell::from_f64(a[0] + b[0]); Some(&self.out) } } @@ -647,7 +645,7 @@ mod tests { /// Mode B barrier join: a 2-input f64 sum that fires only when both inputs /// share the current cycle timestamp (both warm by construction when it fires). struct BarrierSum { - out: [Scalar; 1], + out: [Cell; 1], } impl BarrierSum { fn sig() -> NodeSchema { @@ -662,8 +660,8 @@ mod tests { fn lookbacks(&self) -> Vec { vec![1, 1] } - fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> { - self.out[0] = Scalar::f64(ctx.f64_in(0)[0] + ctx.f64_in(1)[0]); + fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> { + self.out[0] = Cell::from_f64(ctx.f64_in(0)[0] + ctx.f64_in(1)[0]); Some(&self.out) } } @@ -672,7 +670,7 @@ mod tests { /// (input 2). Fires when the pair completes (holding input 2) OR when input 2 /// ticks (holding the pair) — the OR-combine. struct MixedSum { - out: [Scalar; 1], + out: [Cell; 1], } impl MixedSum { fn sig() -> NodeSchema { @@ -691,14 +689,14 @@ mod tests { fn lookbacks(&self) -> Vec { vec![1, 1, 1] } - fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> { + fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> { let a = ctx.f64_in(0); let b = ctx.f64_in(1); let c = ctx.f64_in(2); if a.is_empty() || b.is_empty() || c.is_empty() { return None; } - self.out[0] = Scalar::f64(a[0] + b[0] + c[0]); + self.out[0] = Cell::from_f64(a[0] + b[0] + c[0]); Some(&self.out) } } @@ -708,7 +706,7 @@ mod tests { /// isolation. The five inputs are a Barrier(0) group, so the node emits one /// complete bar only when all five share the cycle's timestamp. struct Ohlcv { - out: [Scalar; 5], + out: [Cell; 5], } impl Ohlcv { fn sig() -> NodeSchema { @@ -729,13 +727,13 @@ mod tests { fn lookbacks(&self) -> Vec { vec![1; 5] } - fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> { + fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> { for i in 0..5 { let w = ctx.f64_in(i); if w.is_empty() { return None; // not yet warmed } - self.out[i] = Scalar::f64(w[0]); + self.out[i] = Cell::from_f64(w[0]); } Some(&self.out) // one 5-field record, all fields co-fresh } @@ -746,7 +744,7 @@ mod tests { /// into an f64 slot; field 1 would not). Its `eval` never runs in these tests — /// bootstrap rejects the wiring first. struct TwoField { - out: [Scalar; 2], + out: [Cell; 2], } impl TwoField { fn sig() -> NodeSchema { @@ -764,9 +762,9 @@ mod tests { fn lookbacks(&self) -> Vec { vec![1] } - fn eval(&mut self, _ctx: Ctx<'_>) -> Option<&[Scalar]> { - self.out[0] = Scalar::f64(0.0); - self.out[1] = Scalar::i64(0); + fn eval(&mut self, _ctx: Ctx<'_>) -> Option<&[Cell]> { + self.out[0] = Cell::from_f64(0.0); + self.out[1] = Cell::from_i64(0); Some(&self.out) } } @@ -775,7 +773,7 @@ mod tests { /// (sink side effect) and returns its value as a one-field output the engine /// forwards downstream (producer). Proves the C8 "both" role. struct TapForward { - out: [Scalar; 1], + out: [Cell; 1], tx: mpsc::Sender<(Timestamp, Vec)>, } impl TapForward { @@ -791,14 +789,14 @@ mod tests { fn lookbacks(&self) -> Vec { vec![1] } - fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> { + fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> { let w = ctx.f64_in(0); if w.is_empty() { return None; } let v = w[0]; let _ = self.tx.send((ctx.now(), vec![Scalar::f64(v)])); // sink side effect - self.out[0] = Scalar::f64(v); + self.out[0] = Cell::from_f64(v); Some(&self.out) // producer output: engine forwards it } } @@ -895,7 +893,7 @@ mod tests { let build = |tx| { boot( vec![ - Box::new(AsOfSum { out: [Scalar::f64(0.0)] }), + Box::new(AsOfSum { out: [Cell::from_f64(0.0)] }), Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)), ], vec![ @@ -941,7 +939,7 @@ mod tests { let build = |tx| { boot( vec![ - Box::new(BarrierSum { out: [Scalar::f64(0.0)] }), + Box::new(BarrierSum { out: [Cell::from_f64(0.0)] }), Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)), ], vec![ @@ -990,7 +988,7 @@ mod tests { vec![ Box::new(Sma::new(2)), Box::new(Sma::new(4)), - Box::new(BarrierSum { out: [Scalar::f64(0.0)] }), + Box::new(BarrierSum { out: [Cell::from_f64(0.0)] }), Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)), ], vec![ @@ -1041,7 +1039,7 @@ mod tests { let (tx, rx) = mpsc::channel(); let mut h = boot( vec![ - Box::new(MixedSum { out: [Scalar::f64(0.0)] }), + Box::new(MixedSum { out: [Cell::from_f64(0.0)] }), Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)), ], vec![ @@ -1153,7 +1151,7 @@ mod tests { let (tx, rx) = mpsc::channel(); let mut h = boot( vec![ - Box::new(Ohlcv { out: [Scalar::f64(0.0); 5] }), + Box::new(Ohlcv { out: [Cell::from_f64(0.0); 5] }), Box::new(Recorder::new( &[ScalarKind::F64, ScalarKind::F64, ScalarKind::F64, ScalarKind::F64, ScalarKind::F64], Firing::Any, @@ -1206,7 +1204,7 @@ mod tests { let build = |tx| { boot( vec![ - Box::new(Ohlcv { out: [Scalar::f64(0.0); 5] }), + Box::new(Ohlcv { out: [Cell::from_f64(0.0); 5] }), Box::new(Sub::new()), Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)), ], @@ -1252,7 +1250,7 @@ mod tests { let (tx, rx) = mpsc::channel(); let mut h = boot( vec![ - Box::new(Ohlcv { out: [Scalar::f64(0.0); 5] }), + Box::new(Ohlcv { out: [Cell::from_f64(0.0); 5] }), Box::new(Sub::new()), Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)), ], @@ -1304,7 +1302,7 @@ mod tests { // Sma(1)'s f64 input slot is a per-field kind mismatch -> KindMismatch. (The // mismatch is field-specific: from_field 0 would have matched.) let err = boot( - vec![Box::new(TwoField { out: [Scalar::f64(0.0), Scalar::i64(0)] }), Box::new(Sma::new(1))], + vec![Box::new(TwoField { out: [Cell::from_f64(0.0), Cell::from_i64(0)] }), Box::new(Sma::new(1))], vec![ TwoField::sig(), Sma::builder().schema().clone(), @@ -1377,7 +1375,7 @@ mod tests { let (tx, rx) = mpsc::channel(); let mut h = boot( vec![ - Box::new(Ohlcv { out: [Scalar::f64(0.0); 5] }), + Box::new(Ohlcv { out: [Cell::from_f64(0.0); 5] }), Box::new(Recorder::new( &[ScalarKind::F64, ScalarKind::F64, ScalarKind::F64, ScalarKind::F64, ScalarKind::F64], Firing::Any, @@ -1472,7 +1470,7 @@ mod tests { let (tx_down, rx_down) = mpsc::channel(); let mut h = boot( vec![ - Box::new(TapForward { out: [Scalar::f64(0.0)], tx: tx_tap }), + Box::new(TapForward { out: [Cell::from_f64(0.0)], tx: tx_tap }), Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx_down)), ], vec![ @@ -1611,7 +1609,7 @@ mod tests { let (tx, _rx) = mpsc::channel(); let err = boot( vec![ - Box::new(TwoField { out: [Scalar::f64(0.0), Scalar::i64(0)] }), + Box::new(TwoField { out: [Cell::from_f64(0.0), Cell::from_i64(0)] }), Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)), ], vec![ @@ -1778,7 +1776,7 @@ mod tests { vec![ Box::new(Sma::new(2)), // 0: shared producer (src A) Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, t_any)), // 1: as-of tap of 0 - Box::new(BarrierSum { out: [Scalar::f64(0.0)] }), // 2: SMA(2) + src B (barrier) + Box::new(BarrierSum { out: [Cell::from_f64(0.0)] }), // 2: SMA(2) + src B (barrier) Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, t_bar)), // 3: tap of barrier ], vec![ @@ -1881,6 +1879,89 @@ mod tests { assert_eq!(rx2.try_iter().collect::>(), expected); // deterministic } + /// A producer emitting a fixed two-field record with DISTINCT non-zero values of + /// two different kinds (field 0 f64 = 2.5, field 1 i64 = 42). The values are + /// chosen so each kind's bit pattern differs from its cross-kind reinterpretation + /// — 42_i64's bits read as an f64 are a denormal ~2e-322, and 2.5_f64's bits read + /// as an i64 are 0x4004000000000000 — so a kind-blind forward is observable. + struct MixedKindEmit { + out: [Cell; 2], + } + impl MixedKindEmit { + fn sig() -> NodeSchema { + NodeSchema { + inputs: vec![f64_port(Firing::Any)], + output: vec![ + FieldSpec { name: "f".into(), kind: ScalarKind::F64 }, + FieldSpec { name: "i".into(), kind: ScalarKind::I64 }, + ], + params: vec![], + } + } + } + impl Node for MixedKindEmit { + fn lookbacks(&self) -> Vec { + vec![1] + } + fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> { + if ctx.f64_in(0).is_empty() { + return None; + } + Some(&self.out) + } + } + + #[test] + fn mixed_kind_record_forwards_each_field_by_its_kind_end_to_end() { + // PROPERTY: the inter-node forward (push_cell) reinterprets each forwarded + // field by the DESTINATION column's kind, so a multi-kind record's i64 field + // survives the tag-free Cell carrier intact alongside its f64 field. Every + // other end-to-end forward test wires f64-only edges; since a Cell is one + // untyped 8-byte word, an i64 field forwarded as if it were f64 (or vice + // versa) would corrupt silently and still pass all f64-only tests — this + // pins the i64 path with values whose bit patterns differ across kinds. + let build = |tx| { + boot( + vec![ + Box::new(MixedKindEmit { out: [Cell::from_f64(2.5), Cell::from_i64(42)] }), + Box::new(Recorder::new(&[ScalarKind::F64, ScalarKind::I64], Firing::Any, tx)), + ], + vec![ + MixedKindEmit::sig(), + recorder_sig(&[ScalarKind::F64, ScalarKind::I64], Firing::Any), + ], + vec![SourceSpec { + kind: ScalarKind::F64, + targets: vec![Target { node: 0, slot: 0 }], + }], + vec![ + Edge { from: 0, to: 1, slot: 0, from_field: 0 }, // f64 field -> f64 slot + Edge { from: 0, to: 1, slot: 1, from_field: 1 }, // i64 field -> i64 slot + ], + ) + .expect("kind-matched mixed-field wiring bootstraps") + }; + let prices = f64_stream(&[(1, 5.0), (2, 7.0)]); + + let (tx, rx) = mpsc::channel(); + let mut h = build(tx); + h.run(vec![Box::new(VecSource::new(prices.clone()))]); + let out: Vec<(Timestamp, Vec)> = rx.try_iter().collect(); + // each cycle the producer fires and emits its constant record; the recorder + // reads field 0 back as f64 2.5 and field 1 back as i64 42 — a kind-blind + // forward would surface a denormal f64 / a giant i64 instead. + let expected = vec![ + (Timestamp(1), vec![Scalar::f64(2.5), Scalar::i64(42)]), + (Timestamp(2), vec![Scalar::f64(2.5), Scalar::i64(42)]), + ]; + assert_eq!(out, expected); + + let (tx2, rx2) = mpsc::channel(); + let mut h2 = build(tx2); + h2.run(vec![Box::new(VecSource::new(prices))]); + assert_eq!(rx2.try_iter().collect::>(), expected); // deterministic (C1) + } + #[test] fn wide_parallel_layer_multi_sink_records_each_stream() { // PROPERTY: one source fanned to several PARALLEL producers of different diff --git a/crates/aura-std/src/add.rs b/crates/aura-std/src/add.rs index 84d8031..f8e9c49 100644 --- a/crates/aura-std/src/add.rs +++ b/crates/aura-std/src/add.rs @@ -2,7 +2,7 @@ //! Combines two signal streams into one — the most basic combinator for the //! north-star "combine one signal with another" research move (C10). -use aura_core::{Ctx, FieldSpec, Firing, Node, NodeSchema, PortSpec, PrimitiveBuilder, Scalar, ScalarKind}; +use aura_core::{Cell, Ctx, FieldSpec, Firing, Node, NodeSchema, PortSpec, PrimitiveBuilder, ScalarKind}; /// Two-input f64 sum: input 0 plus input 1. Emits `None` until both inputs /// have a value. @@ -17,13 +17,13 @@ use aura_core::{Ctx, FieldSpec, Firing, Node, NodeSchema, PortSpec, PrimitiveBui /// two distinct cycles, C4), a fired node emits one row per *cycle*, so a /// recorded combined stream may carry more than one row per timestamp. pub struct Add { - out: [Scalar; 1], + out: [Cell; 1], } impl Add { /// Build an `Add` node. pub fn new() -> Self { - Self { out: [Scalar::f64(0.0)] } + Self { out: [Cell::from_f64(0.0)] } } /// The param-generic recipe for a blueprint primitive: paramless, builds through @@ -55,13 +55,13 @@ impl Node for Add { vec![1, 1] } - fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> { + fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> { let a = ctx.f64_in(0); let b = ctx.f64_in(1); if a.is_empty() || b.is_empty() { return None; } - self.out[0] = Scalar::f64(a[0] + b[0]); + self.out[0] = Cell::from_f64(a[0] + b[0]); Some(&self.out) } @@ -73,7 +73,7 @@ impl Node for Add { #[cfg(test)] mod tests { use super::*; - use aura_core::{AnyColumn, Timestamp}; + use aura_core::{AnyColumn, Scalar, Timestamp}; #[test] fn add_is_sum_once_both_inputs_present() { @@ -89,7 +89,7 @@ mod tests { // both present -> a + b inputs[1].push(Scalar::f64(4.0)).unwrap(); - assert_eq!(add.eval(Ctx::new(&inputs, Timestamp(0))), Some([Scalar::f64(14.0)].as_slice())); + assert_eq!(add.eval(Ctx::new(&inputs, Timestamp(0))), Some([Cell::from_f64(14.0)].as_slice())); } #[test] diff --git a/crates/aura-std/src/ema.rs b/crates/aura-std/src/ema.rs index 0d75d97..93c2342 100644 --- a/crates/aura-std/src/ema.rs +++ b/crates/aura-std/src/ema.rs @@ -18,7 +18,7 @@ //! nothing on the hot path (the output buffer is reused). use aura_core::{ - Ctx, FieldSpec, Firing, Node, NodeSchema, ParamSpec, PortSpec, PrimitiveBuilder, Scalar, + Cell, Ctx, FieldSpec, Firing, Node, NodeSchema, ParamSpec, PortSpec, PrimitiveBuilder, ScalarKind, }; @@ -34,7 +34,7 @@ pub struct Ema { warmup_sum: f64, count: usize, ema: f64, - out: [Scalar; 1], + out: [Cell; 1], } impl Ema { @@ -49,7 +49,7 @@ impl Ema { warmup_sum: 0.0, count: 0, ema: 0.0, - out: [Scalar::f64(0.0)], + out: [Cell::from_f64(0.0)], } } @@ -76,7 +76,7 @@ impl Node for Ema { vec![1] } - fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> { + fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> { let w = ctx.f64_in(0); if w.is_empty() { return None; // no sample yet @@ -96,7 +96,7 @@ impl Node for Ema { // O(1) recurrence: one sub, one mul, one add. self.ema += self.alpha * (x - self.ema); } - self.out[0] = Scalar::f64(self.ema); + self.out[0] = Cell::from_f64(self.ema); Some(&self.out) } @@ -108,7 +108,7 @@ impl Node for Ema { #[cfg(test)] mod tests { use super::*; - use aura_core::{AnyColumn, Timestamp}; + use aura_core::{AnyColumn, Scalar, Timestamp}; #[test] fn ema_warms_up_like_sma_then_smooths() { @@ -128,7 +128,7 @@ mod tests { let got = ema.eval(Ctx::new(&inputs, Timestamp(0))); match want { None => assert_eq!(got, None), - Some(m) => assert_eq!(got, Some([Scalar::f64(m)].as_slice())), + Some(m) => assert_eq!(got, Some([Cell::from_f64(m)].as_slice())), } } } @@ -141,10 +141,10 @@ mod tests { let mut inputs = vec![AnyColumn::with_capacity(ScalarKind::F64, 1)]; inputs[0].push(Scalar::f64(7.0)).unwrap(); - assert_eq!(ema.eval(Ctx::new(&inputs, Timestamp(0))), Some([Scalar::f64(7.0)].as_slice())); + assert_eq!(ema.eval(Ctx::new(&inputs, Timestamp(0))), Some([Cell::from_f64(7.0)].as_slice())); inputs[0].push(Scalar::f64(9.0)).unwrap(); - assert_eq!(ema.eval(Ctx::new(&inputs, Timestamp(0))), Some([Scalar::f64(9.0)].as_slice())); + assert_eq!(ema.eval(Ctx::new(&inputs, Timestamp(0))), Some([Cell::from_f64(9.0)].as_slice())); } #[test] diff --git a/crates/aura-std/src/exposure.rs b/crates/aura-std/src/exposure.rs index d41d709..1dd81a4 100644 --- a/crates/aura-std/src/exposure.rs +++ b/crates/aura-std/src/exposure.rs @@ -4,7 +4,7 @@ //! `scale` sets which signal magnitude maps to full exposure (sizing lives here). use aura_core::{ - Ctx, FieldSpec, Firing, Node, NodeSchema, ParamSpec, PortSpec, PrimitiveBuilder, Scalar, + Cell, Ctx, FieldSpec, Firing, Node, NodeSchema, ParamSpec, PortSpec, PrimitiveBuilder, ScalarKind, }; @@ -12,14 +12,14 @@ use aura_core::{ /// Emits `None` until its input is present (warm-up filter, C8). pub struct Exposure { scale: f64, - out: [Scalar; 1], + out: [Cell; 1], } impl Exposure { /// Build an exposure node with saturation magnitude `scale` (must be > 0). pub fn new(scale: f64) -> Self { assert!(scale > 0.0, "Exposure scale must be > 0"); - Self { scale, out: [Scalar::f64(0.0)] } + Self { scale, out: [Cell::from_f64(0.0)] } } /// The param-generic recipe for a blueprint primitive: declares `scale` and builds @@ -43,12 +43,12 @@ impl Node for Exposure { vec![1] } - fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> { + fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> { let w = ctx.f64_in(0); if w.is_empty() { return None; // not yet warmed up (C8 filter) } - self.out[0] = Scalar::f64((w[0] / self.scale).clamp(-1.0, 1.0)); + self.out[0] = Cell::from_f64((w[0] / self.scale).clamp(-1.0, 1.0)); Some(&self.out) } @@ -60,7 +60,7 @@ impl Node for Exposure { #[cfg(test)] mod tests { use super::*; - use aura_core::{AnyColumn, Timestamp}; + use aura_core::{AnyColumn, Scalar, Timestamp}; #[test] fn exposure_clamps_to_unit_band() { @@ -78,7 +78,7 @@ mod tests { inputs[0].push(Scalar::f64(sig)).unwrap(); assert_eq!( e.eval(Ctx::new(&inputs, Timestamp(0))), - Some([Scalar::f64(want)].as_slice()) + Some([Cell::from_f64(want)].as_slice()) ); } } diff --git a/crates/aura-std/src/lincomb.rs b/crates/aura-std/src/lincomb.rs index 6ecc3da..9a99c66 100644 --- a/crates/aura-std/src/lincomb.rs +++ b/crates/aura-std/src/lincomb.rs @@ -7,7 +7,7 @@ //! indexed `F64` knobs that `Composite::param_space` aggregates (C8/C12/C19). use aura_core::{ - Ctx, FieldSpec, Firing, Node, NodeSchema, ParamSpec, PortSpec, PrimitiveBuilder, Scalar, + Cell, Ctx, FieldSpec, Firing, Node, NodeSchema, ParamSpec, PortSpec, PrimitiveBuilder, ScalarKind, }; @@ -27,7 +27,7 @@ use aura_core::{ /// so a recorded combined stream may carry more than one row per timestamp. pub struct LinComb { weights: Vec, - out: [Scalar; 1], + out: [Cell; 1], } impl LinComb { @@ -37,7 +37,7 @@ impl LinComb { /// Panics if `weights` is empty. pub fn new(weights: Vec) -> Self { assert!(!weights.is_empty(), "LinComb needs at least one weight"); - Self { weights, out: [Scalar::f64(0.0)] } + Self { weights, out: [Cell::from_f64(0.0)] } } /// The param-generic recipe for a blueprint primitive. The `arity` is topology @@ -65,7 +65,7 @@ impl Node for LinComb { vec![1; self.weights.len()] } - fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> { + fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> { let mut acc = 0.0; for (i, &w) in self.weights.iter().enumerate() { let w_in = ctx.f64_in(i); @@ -74,7 +74,7 @@ impl Node for LinComb { } acc += w * w_in[0]; } - self.out[0] = Scalar::f64(acc); + self.out[0] = Cell::from_f64(acc); Some(&self.out) } @@ -86,7 +86,7 @@ impl Node for LinComb { #[cfg(test)] mod tests { use super::*; - use aura_core::{AnyColumn, Timestamp}; + use aura_core::{AnyColumn, Scalar, Timestamp}; #[test] fn lincomb_weighted_sum_once_all_present() { @@ -102,7 +102,7 @@ mod tests { // both present -> 0.5*10 + 2.0*3 = 11.0 inputs[1].push(Scalar::f64(3.0)).unwrap(); - assert_eq!(lc.eval(Ctx::new(&inputs, Timestamp(0))), Some([Scalar::f64(11.0)].as_slice())); + assert_eq!(lc.eval(Ctx::new(&inputs, Timestamp(0))), Some([Cell::from_f64(11.0)].as_slice())); } #[test] @@ -115,7 +115,7 @@ mod tests { inputs[0].push(Scalar::f64(7.0)).unwrap(); inputs[1].push(Scalar::f64(5.0)).unwrap(); // unit weights reproduce Add: 7 + 5 - assert_eq!(lc.eval(Ctx::new(&inputs, Timestamp(0))), Some([Scalar::f64(12.0)].as_slice())); + assert_eq!(lc.eval(Ctx::new(&inputs, Timestamp(0))), Some([Cell::from_f64(12.0)].as_slice())); } #[test] @@ -133,7 +133,7 @@ mod tests { inputs[2].push(Scalar::f64(3.0)).unwrap(); // all warm -> 1 + 2 + 3 - assert_eq!(lc.eval(Ctx::new(&inputs, Timestamp(0))), Some([Scalar::f64(6.0)].as_slice())); + assert_eq!(lc.eval(Ctx::new(&inputs, Timestamp(0))), Some([Cell::from_f64(6.0)].as_slice())); } #[test] @@ -165,7 +165,7 @@ mod tests { inputs[1].push(Scalar::f64(3.0)).unwrap(); // 0.5*10 + 2.0*3 = 11.0 — holds ONLY if each weight landed in its right slot // (a swap would give 2.0*10 + 0.5*3 = 21.5) - assert_eq!(lc.eval(Ctx::new(&inputs, Timestamp(0))), Some([Scalar::f64(11.0)].as_slice())); + assert_eq!(lc.eval(Ctx::new(&inputs, Timestamp(0))), Some([Cell::from_f64(11.0)].as_slice())); // partial: bind weights[0], leave weights[1] open → inject it at build let partial = LinComb::builder(2).bind("weights[0]", Scalar::f64(0.5)); @@ -180,6 +180,6 @@ mod tests { ]; inputs2[0].push(Scalar::f64(10.0)).unwrap(); inputs2[1].push(Scalar::f64(3.0)).unwrap(); - assert_eq!(lc2.eval(Ctx::new(&inputs2, Timestamp(0))), Some([Scalar::f64(11.0)].as_slice())); + assert_eq!(lc2.eval(Ctx::new(&inputs2, Timestamp(0))), Some([Cell::from_f64(11.0)].as_slice())); } } diff --git a/crates/aura-std/src/recorder.rs b/crates/aura-std/src/recorder.rs index 36571b7..c81729d 100644 --- a/crates/aura-std/src/recorder.rs +++ b/crates/aura-std/src/recorder.rs @@ -7,7 +7,7 @@ //! four base scalar kinds so any column can be persisted; returns `None` (filters) //! until every input column is warm. -use aura_core::{Ctx, Firing, Node, NodeSchema, PortSpec, PrimitiveBuilder, Scalar, ScalarKind, Timestamp}; +use aura_core::{Cell, Ctx, Firing, Node, NodeSchema, PortSpec, PrimitiveBuilder, Scalar, ScalarKind, Timestamp}; use std::sync::mpsc::Sender; /// A recording sink over `kinds.len()` input columns. Each fired cycle it reads @@ -57,7 +57,7 @@ impl Node for Recorder { vec![1; self.kinds.len()] } - fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> { + fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> { let mut row = Vec::with_capacity(self.kinds.len()); for (i, &kind) in self.kinds.iter().enumerate() { // newest of each column by kind; `?` returns None (warm-up) if cold. diff --git a/crates/aura-std/src/sim_broker.rs b/crates/aura-std/src/sim_broker.rs index 9931c9d..6deb8de 100644 --- a/crates/aura-std/src/sim_broker.rs +++ b/crates/aura-std/src/sim_broker.rs @@ -4,7 +4,7 @@ //! each cycle (decided at t-1) into a cumulative synthetic pip-equity output. //! Measures signal quality, not execution-modelled P&L. -use aura_core::{Ctx, FieldSpec, Firing, Node, NodeSchema, PortSpec, PrimitiveBuilder, Scalar, ScalarKind}; +use aura_core::{Cell, Ctx, FieldSpec, Firing, Node, NodeSchema, PortSpec, PrimitiveBuilder, ScalarKind}; /// Integrates `exposure * price-return` into cumulative pips. `pip_size` is /// per-instrument reference metadata (beside the hot path, C7/C15), held here, @@ -38,7 +38,7 @@ pub struct SimBroker { prev_price: Option, prev_exposure: f64, // exposure held into this cycle (decided at t-1); 0.0 = flat cum: f64, // cumulative pips - out: [Scalar; 1], + out: [Cell; 1], } impl SimBroker { @@ -51,7 +51,7 @@ impl SimBroker { prev_price: None, prev_exposure: 0.0, cum: 0.0, - out: [Scalar::f64(0.0)], + out: [Cell::from_f64(0.0)], } } @@ -79,7 +79,7 @@ impl Node for SimBroker { vec![1, 1] } - fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> { + fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> { let price = ctx.f64_in(1); if price.is_empty() { return None; // no price yet — nothing to mark @@ -91,7 +91,7 @@ impl Node for SimBroker { } self.prev_price = Some(price); self.prev_exposure = expo; // update AFTER taking PnL — no look-ahead (C2) - self.out[0] = Scalar::f64(self.cum); + self.out[0] = Cell::from_f64(self.cum); Some(&self.out) } @@ -103,7 +103,7 @@ impl Node for SimBroker { #[cfg(test)] mod tests { use super::*; - use aura_core::{AnyColumn, Timestamp}; + use aura_core::{AnyColumn, Scalar, Timestamp}; fn two_f64_inputs() -> Vec { vec![ @@ -121,7 +121,7 @@ mod tests { } inputs[1].push(Scalar::f64(price)).unwrap(); match b.eval(Ctx::new(inputs, Timestamp(0))) { - Some([s]) => s.as_f64(), + Some([c]) => c.f64(), other => panic!("expected Some([F64]), got {other:?}"), } } diff --git a/crates/aura-std/src/sma.rs b/crates/aura-std/src/sma.rs index a76909c..c95a377 100644 --- a/crates/aura-std/src/sma.rs +++ b/crates/aura-std/src/sma.rs @@ -4,21 +4,21 @@ //! engine present (the test drives it by hand, as the sim loop later will). use aura_core::{ - Ctx, FieldSpec, Firing, Node, NodeSchema, ParamSpec, PortSpec, PrimitiveBuilder, Scalar, + Cell, Ctx, FieldSpec, Firing, Node, NodeSchema, ParamSpec, PortSpec, PrimitiveBuilder, ScalarKind, }; /// Simple moving average over the last `length` values of one f64 input. pub struct Sma { length: usize, - out: [Scalar; 1], + out: [Cell; 1], } impl Sma { /// Build an SMA of window `length` (must be >= 1). pub fn new(length: usize) -> Self { assert!(length >= 1, "SMA length must be >= 1"); - Self { length, out: [Scalar::f64(0.0)] } + Self { length, out: [Cell::from_f64(0.0)] } } /// The param-generic recipe for a blueprint primitive: declares `length` and builds @@ -42,7 +42,7 @@ impl Node for Sma { vec![self.length] } - fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> { + fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> { let w = ctx.f64_in(0); if w.len() < self.length { return None; // not yet warmed up @@ -51,7 +51,7 @@ impl Node for Sma { for k in 0..self.length { sum += w[k]; // index 0 = newest (financial indexing) } - self.out[0] = Scalar::f64(sum / self.length as f64); + self.out[0] = Cell::from_f64(sum / self.length as f64); Some(&self.out) } @@ -63,7 +63,7 @@ impl Node for Sma { #[cfg(test)] mod tests { use super::*; - use aura_core::{AnyColumn, Timestamp}; + use aura_core::{AnyColumn, Scalar, Timestamp}; #[test] fn sma_warms_up_then_tracks_the_window_mean() { @@ -83,7 +83,7 @@ mod tests { let got = sma.eval(Ctx::new(&inputs, Timestamp(0))); match want { None => assert_eq!(got, None), - Some(m) => assert_eq!(got, Some([Scalar::f64(m)].as_slice())), + Some(m) => assert_eq!(got, Some([Cell::from_f64(m)].as_slice())), } } } @@ -94,10 +94,10 @@ mod tests { let mut inputs = vec![AnyColumn::with_capacity(ScalarKind::F64, 1)]; inputs[0].push(Scalar::f64(7.0)).unwrap(); - assert_eq!(sma.eval(Ctx::new(&inputs, Timestamp(0))), Some([Scalar::f64(7.0)].as_slice())); + assert_eq!(sma.eval(Ctx::new(&inputs, Timestamp(0))), Some([Cell::from_f64(7.0)].as_slice())); inputs[0].push(Scalar::f64(9.0)).unwrap(); - assert_eq!(sma.eval(Ctx::new(&inputs, Timestamp(0))), Some([Scalar::f64(9.0)].as_slice())); + assert_eq!(sma.eval(Ctx::new(&inputs, Timestamp(0))), Some([Cell::from_f64(9.0)].as_slice())); } #[test] diff --git a/crates/aura-std/src/sub.rs b/crates/aura-std/src/sub.rs index eaae6d1..72254b6 100644 --- a/crates/aura-std/src/sub.rs +++ b/crates/aura-std/src/sub.rs @@ -3,18 +3,18 @@ //! real fan-out + join to run (two SMAs joining into one node), exercising //! multi-input `Ctx` access inside a running graph. -use aura_core::{Ctx, FieldSpec, Firing, Node, NodeSchema, PortSpec, PrimitiveBuilder, Scalar, ScalarKind}; +use aura_core::{Cell, Ctx, FieldSpec, Firing, Node, NodeSchema, PortSpec, PrimitiveBuilder, ScalarKind}; /// Two-input f64 difference: input 0 minus input 1. Emits `None` until both /// inputs have a value. pub struct Sub { - out: [Scalar; 1], + out: [Cell; 1], } impl Sub { /// Build a `Sub` node. pub fn new() -> Self { - Self { out: [Scalar::f64(0.0)] } + Self { out: [Cell::from_f64(0.0)] } } /// The param-generic recipe for a blueprint primitive: paramless, builds through @@ -46,13 +46,13 @@ impl Node for Sub { vec![1, 1] } - fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> { + fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> { let a = ctx.f64_in(0); let b = ctx.f64_in(1); if a.is_empty() || b.is_empty() { return None; } - self.out[0] = Scalar::f64(a[0] - b[0]); + self.out[0] = Cell::from_f64(a[0] - b[0]); Some(&self.out) } @@ -64,7 +64,7 @@ impl Node for Sub { #[cfg(test)] mod tests { use super::*; - use aura_core::{AnyColumn, Timestamp}; + use aura_core::{AnyColumn, Scalar, Timestamp}; #[test] fn sub_is_difference_once_both_inputs_present() { @@ -80,7 +80,7 @@ mod tests { // both present -> a - b inputs[1].push(Scalar::f64(4.0)).unwrap(); - assert_eq!(sub.eval(Ctx::new(&inputs, Timestamp(0))), Some([Scalar::f64(6.0)].as_slice())); + assert_eq!(sub.eval(Ctx::new(&inputs, Timestamp(0))), Some([Cell::from_f64(6.0)].as_slice())); } #[test] diff --git a/docs/design/INDEX.md b/docs/design/INDEX.md index f4592da..8f3bcfe 100644 --- a/docs/design/INDEX.md +++ b/docs/design/INDEX.md @@ -204,10 +204,22 @@ vs. the 16-byte tagged enum). `Scalar` becomes `{ kind: ScalarKind, cell: Cell } rendering) — with `debug_assert`-guarded native accessors (caller asserts the kind; free in release) and a hand-written **value** `PartialEq` that preserves the former enum's IEEE-754 semantics (`NaN != NaN`, `+0.0 == -0.0`), pinned by -the `scalar_eq_is_value_not_bitwise` fixture. Behaviour-preserving. `Cell` is -**not yet** the hot-path carrier — `eval -> Option<&[Scalar]>` and the edges -still flow `Scalar`; replacing the carrier with `Cell` is a larger step that will -get its own note (and touch C8's `eval` contract). +the `scalar_eq_is_value_not_bitwise` fixture. Behaviour-preserving. + +**Realization (`Cell` becomes the hot-path carrier, 2026-06, #74).** The carrier +swap deferred above has landed: `Node::eval` now returns `Option<&[Cell]>` and +every node out-buffer is `[Cell; N]`, so the inter-node forward carries tag-free +8-byte words. The kind lives only at the schema/column: the harness forwards each +field via the new branch-free `AnyColumn::push_cell` (infallible — the edge +kind match is verified once at bootstrap, the surviving guard +`bootstrap_rejects_*_kind_mismatch`). `Scalar` remains on the self-describing +dynamic boundaries: the param plane (`build`/`bind`/`compile_with_params`/sweep +points/`RunManifest`), `AnyColumn::get` (the type-erased read for sinks/serde), +and source ingestion (`Source::next`, the heterogeneous C3 merge). The removed +per-value runtime kind check on node output is the same authoring-bug class C8 +already leaves to a `debug_assert` (output width); node-output-kind correctness +is the node's declared `FieldSpec` contract, caught by each node's own test. +Behaviour-preserving (C1). ### C8 — The node contract **Guarantee.** A node has a **signature** — its `NodeSchema`: each input's scalar @@ -218,7 +230,7 @@ recipe** (C19), so the whole interface is legible without building (cycle 0024). The built node implements `lookbacks()` — the per-input buffer depth, the one quantity that may depend on an injected param (e.g. an `Sma`'s window = its `length`), read once at bootstrap to size the windows — and `eval(ctx) -> -Option<&[Scalar]>`. The +Option<&[Cell]>`. The engine provides read-only, zero-copy windows into each input's SoA ring buffer (`ctx.f64_in(x)[k]`, sized at wiring); a node may *additionally* keep its own mutable series for derived/intermediate state. `None`/Void return = filter / @@ -239,8 +251,9 @@ columns; length 1 = scalar). Binding is **field-wise only**: `Edge::from_field` selects one producer column per edge; consuming a whole record is N edges (no "bind whole record" mechanism). The K fields of one record are **co-fresh by construction** (one `eval`, one timestamp), so C6 is untouched. `eval` returns -`Option<&[Scalar]>` — a borrowed row into a node-owned buffer — so the forward -path allocates nothing per cycle (C7). +`Option<&[Cell]>` — a borrowed row into a node-owned buffer — so the forward +path allocates nothing per cycle (C7) (the carrier is now a tag-free `Cell` — see +the C7 carrier note). **Realization (cycle 0006).** The pure-consumer (sink) half of this contract is now realized at the substrate: **recording is a node role, not a type.** A recording node reads its typed input windows + `ctx.now()` in `eval` and pushes