From 1100a60c7672f9eeedd6790ccaa8a5f8d4843eb9 Mon Sep 17 00:00:00 2001 From: Brummel Date: Thu, 4 Jun 2026 14:38:26 +0200 Subject: [PATCH] feat: recording is a node role, not a type (multi-sink substrate) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace the engine's single observe: usize recording affordance with recording-by-node, so one run records many streams. A recording node reads its typed input windows + ctx.now() in eval and pushes the record to a destination it holds as a field (a channel, a chart handle) — an out-of-graph side effect. There is no Sink type, trait, or engine flag: a pure-consumer node returns None, and a node may record AND return a forwarded output in the same eval (the C8 "both" case). In-graph routing stays engine-owned data (the edge table); the escape out of the graph is the node's own side effect, and that boundary is the determinism / graph-as-data boundary. Engine surface shrinks: Ctx gains now: Timestamp + now() (C2-causal, the present cycle's timestamp); Harness loses the observe field, its observe >= n bootstrap check, and the per-cycle observed-row collection; bootstrap drops its 4th param; run returns (). Recorded streams are now sparse and timestamped (a record per fired cycle) instead of the dense Vec>. The Ctx::new signature change touched 9 call sites across three crates (not the 3 the spec estimated) — aura-std's node tests and the engine run loop were threaded too. The engine test suite migrated to a test-local Recorder fixture whose read-back is an mpsc channel, never Rc/RefCell, keeping aura-engine/src purity-clean (C7). Eight new proof tests cover the multi-sink headline, producer-and-sink, mixed-kind recording, all-fields tap, both recorder firing modes, determinism, and recorder-edge kind rejection. C8/C22 gain cycle-0006 realization notes. Gates: workspace test 45 green (core 20, std 3, engine 22), clippy -D warnings clean, purity grep clean (only a comment names Rc/RefCell). closes #2 --- crates/aura-core/src/ctx.rs | 32 +- crates/aura-engine/src/harness.rs | 746 +++++++++++++++++++++++------- crates/aura-std/src/sma.rs | 8 +- crates/aura-std/src/sub.rs | 6 +- docs/design/INDEX.md | 19 + 5 files changed, 617 insertions(+), 194 deletions(-) diff --git a/crates/aura-core/src/ctx.rs b/crates/aura-core/src/ctx.rs index aa74192..d855691 100644 --- a/crates/aura-core/src/ctx.rs +++ b/crates/aura-core/src/ctx.rs @@ -6,16 +6,25 @@ use crate::{AnyColumn, Timestamp, Window}; /// Read-only, zero-copy view of a node's inputs for one `eval`, in schema -/// order. `Copy` because it is just a borrow of the input slice. +/// order, plus the cycle's timestamp (C4). `Copy` because it is just a borrow of +/// the input slice plus a `Copy` timestamp. #[derive(Clone, Copy)] pub struct Ctx<'a> { inputs: &'a [AnyColumn], + now: Timestamp, } impl<'a> Ctx<'a> { - /// Wrap the per-input columns (in schema-declared order) for one `eval`. - pub fn new(inputs: &'a [AnyColumn]) -> Self { - Self { inputs } + /// Wrap the per-input columns (in schema-declared order) and the cycle + /// timestamp for one `eval`. + pub fn new(inputs: &'a [AnyColumn], now: Timestamp) -> Self { + Self { inputs, now } + } + + /// The current cycle's timestamp (C4). Causal — the present cycle's + /// timestamp, never the future (C2) — so reading it introduces no look-ahead. + pub fn now(&self) -> Timestamp { + self.now } /// Zero-copy `f64` window into input `i` (index 0 = newest). Panics if input @@ -61,7 +70,7 @@ impl<'a> Ctx<'a> { #[cfg(test)] mod tests { use super::*; - use crate::{Scalar, ScalarKind}; + use crate::{Scalar, ScalarKind, Timestamp}; #[test] fn ctx_hands_financial_indexed_windows() { @@ -69,7 +78,7 @@ mod tests { for v in [10.0_f64, 20.0, 30.0] { inputs[0].push(Scalar::F64(v)).unwrap(); } - let ctx = Ctx::new(&inputs); + let ctx = Ctx::new(&inputs, Timestamp(0)); let w = ctx.f64_in(0); assert_eq!(w.len(), 3); assert_eq!(w[0], 30.0); // newest @@ -84,7 +93,7 @@ mod tests { ]; inputs[0].push(Scalar::F64(1.5)).unwrap(); inputs[1].push(Scalar::I64(42)).unwrap(); - let ctx = Ctx::new(&inputs); + let ctx = Ctx::new(&inputs, Timestamp(0)); assert_eq!(ctx.f64_in(0)[0], 1.5); assert_eq!(ctx.i64_in(1)[0], 42); } @@ -94,7 +103,14 @@ mod tests { fn ctx_panics_on_kind_mismatch() { let mut inputs = vec![AnyColumn::with_capacity(ScalarKind::I64, 2)]; inputs[0].push(Scalar::I64(7)).unwrap(); - let ctx = Ctx::new(&inputs); + let ctx = Ctx::new(&inputs, Timestamp(0)); let _ = ctx.f64_in(0); // wrong kind → panic } + + #[test] + fn ctx_now_returns_cycle_timestamp() { + let inputs: Vec = vec![]; + let ctx = Ctx::new(&inputs, Timestamp(42)); + assert_eq!(ctx.now(), Timestamp(42)); + } } diff --git a/crates/aura-engine/src/harness.rs b/crates/aura-engine/src/harness.rs index 25e16fc..39e03e0 100644 --- a/crates/aura-engine/src/harness.rs +++ b/crates/aura-engine/src/harness.rs @@ -5,7 +5,8 @@ //! the flat, monomorphized sharpening of RustAst's reference-counted, //! interior-mutable observer push graph: no reference counting, no interior //! mutability, no per-cycle allocation (C1/C7). Each node owns its input columns -//! (the cycle-0002 shape), so `Ctx` is unchanged. A node's `eval` returns a +//! (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 //! (`Edge::from_field`) into a consumer slot, so the K fields are co-fresh (C6). //! @@ -56,7 +57,7 @@ pub struct SourceSpec { pub enum BootstrapError { /// An edge or source-target connects mismatched scalar kinds. KindMismatch { producer: ScalarKind, consumer: ScalarKind }, - /// A node or slot index in an edge or target (or the observe index) is out of range. + /// A node or slot index in an edge or target is out of range. BadIndex, /// The wiring contains a directed cycle (only an explicit delay node may close /// a loop; that node does not exist yet). @@ -88,7 +89,6 @@ pub struct Harness { topo: Vec, out_edges: Vec>, sources: Vec, - observe: usize, } // Manual, node-opaque `Debug`: `Box` is not `Debug`, so the struct @@ -102,7 +102,6 @@ impl core::fmt::Debug for Harness { .field("topo", &self.topo) .field("out_edges", &self.out_edges) .field("sources", &self.sources) - .field("observe", &self.observe) .finish() } } @@ -116,12 +115,8 @@ impl Harness { nodes: Vec>, sources: Vec, edges: Vec, - observe: usize, ) -> Result { let n = nodes.len(); - if observe >= n { - return Err(BootstrapError::BadIndex); - } let schemas: Vec<_> = nodes.iter().map(|nd| nd.schema()).collect(); @@ -201,16 +196,16 @@ impl Harness { topo, out_edges, sources, - observe, }) } /// Drive the sources, k-way-merged in timestamp order (ties by source index, - /// C4); returns the observed node's per-cycle emission (`Some` when it fired - /// and produced output, `None` when it held or filtered). One stream per - /// source, each ascending in timestamp (C3 ingestion precondition). Allocates - /// nothing per cycle beyond the output vector. - pub fn run(&mut self, streams: Vec>) -> Vec>> { + /// C4). One stream per source, each ascending in timestamp (C3 ingestion + /// precondition). Recording is a node-side concern: a recording node pushes + /// its record to a destination it holds (out of graph) inside `eval`; the + /// engine only routes in-graph edges and is oblivious to the side effect. + /// Allocates nothing per cycle beyond the reused scratch buffer. + pub fn run(&mut self, streams: Vec>) { assert_eq!( streams.len(), self.sources.len(), @@ -221,12 +216,10 @@ impl Harness { // disjoint field borrows so the topo walk can read topo/out_edges/sources // while mutating nodes - let Harness { nodes, topo, out_edges, sources, observe } = self; - let observe = *observe; + let Harness { nodes, topo, out_edges, sources } = self; let mut cursor: Vec = vec![0; streams.len()]; let mut cycle_id: u64 = 0; - let mut out = Vec::new(); let mut scratch: Vec = Vec::new(); loop { @@ -260,7 +253,6 @@ impl Harness { } // evaluate in topological order; gate by firing; forward Some outputs - let mut observed: Option> = None; for &nidx in topo.iter() { let out_len = nodes[nidx].out_len; let fired = { @@ -272,11 +264,8 @@ impl Harness { } let result: Option<&[Scalar]> = { let nb = &mut nodes[nidx]; - nb.node.eval(Ctx::new(&nb.inputs)) + nb.node.eval(Ctx::new(&nb.inputs, ts)) }; - if nidx == observe { - observed = result.map(|row| row.to_vec()); - } if let Some(row) = result { debug_assert_eq!(row.len(), out_len, "node returned a row of the wrong width"); scratch.clear(); @@ -290,9 +279,7 @@ impl Harness { } } } - out.push(observed); } - out } } @@ -351,6 +338,7 @@ mod tests { // brought in by `use super::*` — the fixtures construct them, so import here. use aura_core::{FieldSpec, InputSpec, NodeSchema}; use aura_std::{Sma, Sub}; + use std::sync::mpsc; /// Build an f64 source stream from (timestamp, value) points. fn f64_stream(points: &[(i64, f64)]) -> Vec<(Timestamp, Scalar)> { @@ -498,215 +486,342 @@ mod tests { } } + /// A recording node (test-local fixture; stands in for a downstream author's + /// chart/registry sink). It declares typed input slots and holds an + /// `mpsc::Sender`; on every fired cycle it reads the newest of each input plus + /// `ctx.now()`, sends the timestamped record out of the graph, and returns + /// `None` (pure consumer — C8). Read-back is via the channel, never `Rc`/ + /// `RefCell`, so `aura-engine/src` stays free of the interior-mutability the + /// purity invariant (C7) forbids. + struct Recorder { + kinds: Vec, + firing: Firing, + tx: mpsc::Sender<(Timestamp, Vec)>, + } + impl Recorder { + fn new( + kinds: &[ScalarKind], + firing: Firing, + tx: mpsc::Sender<(Timestamp, Vec)>, + ) -> Self { + Self { kinds: kinds.to_vec(), firing, tx } + } + } + impl Node for Recorder { + fn schema(&self) -> NodeSchema { + NodeSchema { + inputs: self + .kinds + .iter() + .map(|&kind| InputSpec { kind, lookback: 1, firing: self.firing }) + .collect(), + output: vec![], // pure sink: no output port + } + } + fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> { + let mut row = Vec::with_capacity(self.kinds.len()); + for (i, &kind) in self.kinds.iter().enumerate() { + let v = match kind { + ScalarKind::I64 => { + let w = ctx.i64_in(i); + if w.is_empty() { + return None; // not yet warmed + } + Scalar::I64(w[0]) + } + ScalarKind::F64 => { + let w = ctx.f64_in(i); + if w.is_empty() { + return None; + } + Scalar::F64(w[0]) + } + ScalarKind::Bool => { + let w = ctx.bool_in(i); + if w.is_empty() { + return None; + } + Scalar::Bool(w[0]) + } + ScalarKind::Timestamp => { + let w = ctx.ts_in(i); + if w.is_empty() { + return None; + } + Scalar::Ts(w[0]) + } + }; + row.push(v); + } + let _ = self.tx.send((ctx.now(), row)); // out-of-graph side effect + None // records, forwards nothing + } + } + + /// A node that records AND forwards: it sends `(now, value)` out of the graph + /// (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], + tx: mpsc::Sender<(Timestamp, Vec)>, + } + impl Node for TapForward { + fn schema(&self) -> NodeSchema { + NodeSchema { + inputs: vec![InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Any }], + output: vec![FieldSpec { name: "value", kind: ScalarKind::F64 }], + } + } + fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> { + 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); + Some(&self.out) // producer output: engine forwards it + } + } + #[test] fn chain_source_sma_runs() { - // node 0 = SMA(3); one source -> SMA(3).in0; observe node 0 + // node 0 = SMA(3); source -> SMA(3).in0; node 1 = Recorder taps node 0. + let (tx, rx) = mpsc::channel(); let mut h = Harness::bootstrap( - vec![Box::new(Sma::new(3))], + vec![ + Box::new(Sma::new(3)), + Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)), + ], vec![SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }], - vec![], - 0, + vec![Edge { from: 0, to: 1, slot: 0, from_field: 0 }], ) .expect("valid"); - let out = h.run(vec![f64_stream(&[(1, 1.0), (2, 2.0), (3, 3.0), (4, 4.0), (5, 5.0)])]); + h.run(vec![f64_stream(&[(1, 1.0), (2, 2.0), (3, 3.0), (4, 4.0), (5, 5.0)])]); + let got: Vec<(Timestamp, Vec)> = rx.try_iter().collect(); + // SMA(3) warms at cycle 3; the recorder captures only fired cycles, each + // tagged with the cycle's timestamp (sparse — no None hold-rows). assert_eq!( - out, + got, vec![ - None, - None, - Some(vec![Scalar::F64(2.0)]), - Some(vec![Scalar::F64(3.0)]), - Some(vec![Scalar::F64(4.0)]), + (Timestamp(3), vec![Scalar::F64(2.0)]), + (Timestamp(4), vec![Scalar::F64(3.0)]), + (Timestamp(5), vec![Scalar::F64(4.0)]), ] ); } #[test] fn fan_out_join_dag_runs_deterministically() { - // 0 = SMA(2), 1 = SMA(4), 2 = Sub; one source fans into both SMAs (all - // Any), SMAs join into Sub — the 0003 compat baseline on the new API. - let build = || { + // 0 = SMA(2), 1 = SMA(4), 2 = Sub; source fans into both SMAs; SMAs join + // into Sub; node 3 = Recorder taps Sub — the 0003 baseline on the new API. + let build = |tx| { Harness::bootstrap( - vec![Box::new(Sma::new(2)), Box::new(Sma::new(4)), Box::new(Sub::new())], + vec![ + Box::new(Sma::new(2)), + Box::new(Sma::new(4)), + Box::new(Sub::new()), + Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)), + ], vec![SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }], }], - vec![Edge { from: 0, to: 2, slot: 0, from_field: 0 }, Edge { from: 1, to: 2, slot: 1, from_field: 0 }], - 2, + vec![ + Edge { from: 0, to: 2, slot: 0, from_field: 0 }, + Edge { from: 1, to: 2, slot: 1, from_field: 0 }, + Edge { from: 2, to: 3, slot: 0, from_field: 0 }, + ], ) .expect("valid DAG") }; let prices = f64_stream(&[(1, 10.0), (2, 12.0), (3, 14.0), (4, 16.0), (5, 18.0), (6, 20.0)]); - let mut h = build(); - let out = h.run(vec![prices.clone()]); + let (tx, rx) = mpsc::channel(); + let mut h = build(tx); + h.run(vec![prices.clone()]); + let out: Vec<(Timestamp, Vec)> = rx.try_iter().collect(); // Sub fires once SMA(4) is warm (cycle 4): 15-13, 17-15, 19-17 -> 2. assert_eq!( out, vec![ - None, - None, - None, - Some(vec![Scalar::F64(2.0)]), - Some(vec![Scalar::F64(2.0)]), - Some(vec![Scalar::F64(2.0)]), + (Timestamp(4), vec![Scalar::F64(2.0)]), + (Timestamp(5), vec![Scalar::F64(2.0)]), + (Timestamp(6), vec![Scalar::F64(2.0)]), ] ); - // determinism (C1): a second identical run is bit-identical - let mut h2 = build(); - assert_eq!(h2.run(vec![prices]), out); + // determinism (C1): a second identical run drains a bit-identical stream. + let (tx2, rx2) = mpsc::channel(); + let mut h2 = build(tx2); + h2.run(vec![prices]); + let out2: Vec<(Timestamp, Vec)> = rx2.try_iter().collect(); + assert_eq!(out2, out); } #[test] fn mode_a_as_of_fires_on_any_fresh_and_holds() { - // source 0 ticks t=1,2,3,4; source 1 ticks t=2,4 (slower); both inputs Any. - let build = || { + // AsOfSum @0; node 1 = Recorder taps it. source 0 ticks t=1..4; source 1 + // ticks t=2,4 (slower); both AsOfSum inputs Any. + let build = |tx| { Harness::bootstrap( - vec![Box::new(AsOfSum { out: [Scalar::F64(0.0)] })], + vec![ + Box::new(AsOfSum { out: [Scalar::F64(0.0)] }), + Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)), + ], vec![ SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }, SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 1 }] }, ], - vec![], - 0, + vec![Edge { from: 0, to: 1, slot: 0, from_field: 0 }], ) .expect("valid") }; let s0 = f64_stream(&[(1, 10.0), (2, 20.0), (3, 30.0), (4, 40.0)]); let s1 = f64_stream(&[(2, 100.0), (4, 200.0)]); - let mut h = build(); - let out = h.run(vec![s0.clone(), s1.clone()]); + let (tx, rx) = mpsc::channel(); + let mut h = build(tx); + h.run(vec![s0.clone(), s1.clone()]); + let out: Vec<(Timestamp, Vec)> = rx.try_iter().collect(); // holds s1=100 across t=3 and the t=4 s0-cycle; emits on every tick once warm. assert_eq!( out, vec![ - None, - None, - Some(vec![Scalar::F64(120.0)]), - Some(vec![Scalar::F64(130.0)]), - Some(vec![Scalar::F64(140.0)]), - Some(vec![Scalar::F64(240.0)]), + (Timestamp(2), vec![Scalar::F64(120.0)]), + (Timestamp(3), vec![Scalar::F64(130.0)]), + (Timestamp(4), vec![Scalar::F64(140.0)]), + (Timestamp(4), vec![Scalar::F64(240.0)]), ] ); - let mut h2 = build(); - assert_eq!(h2.run(vec![s0, s1]), out); // deterministic + let (tx2, rx2) = mpsc::channel(); + let mut h2 = build(tx2); + h2.run(vec![s0, s1]); + let out2: Vec<(Timestamp, Vec)> = rx2.try_iter().collect(); + assert_eq!(out2, out); // deterministic } #[test] fn mode_b_barrier_fires_only_on_timestamp_coincidence() { - // identical wiring to mode A, but both inputs are Barrier(0). - let build = || { + // identical wiring to mode A, but both BarrierSum inputs are Barrier(0). + let build = |tx| { Harness::bootstrap( - vec![Box::new(BarrierSum { out: [Scalar::F64(0.0)] })], + vec![ + Box::new(BarrierSum { out: [Scalar::F64(0.0)] }), + Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)), + ], vec![ SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }, SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 1 }] }, ], - vec![], - 0, + vec![Edge { from: 0, to: 1, slot: 0, from_field: 0 }], ) .expect("valid") }; let s0 = f64_stream(&[(1, 10.0), (2, 20.0), (3, 30.0), (4, 40.0)]); let s1 = f64_stream(&[(2, 100.0), (4, 200.0)]); - let mut h = build(); - let out = h.run(vec![s0.clone(), s1.clone()]); - // emits ONLY at t=2 and t=4 where both inputs share the timestamp; holds otherwise. + let (tx, rx) = mpsc::channel(); + let mut h = build(tx); + h.run(vec![s0.clone(), s1.clone()]); + let out: Vec<(Timestamp, Vec)> = rx.try_iter().collect(); + // records ONLY at t=2 and t=4 where both inputs share the timestamp. assert_eq!( out, vec![ - None, - None, - Some(vec![Scalar::F64(120.0)]), - None, - None, - Some(vec![Scalar::F64(240.0)]), + (Timestamp(2), vec![Scalar::F64(120.0)]), + (Timestamp(4), vec![Scalar::F64(240.0)]), ] ); - let mut h2 = build(); - assert_eq!(h2.run(vec![s0, s1]), out); // deterministic + let (tx2, rx2) = mpsc::channel(); + let mut h2 = build(tx2); + h2.run(vec![s0, s1]); + let out2: Vec<(Timestamp, Vec)> = rx2.try_iter().collect(); + assert_eq!(out2, out); // deterministic } #[test] fn within_source_diamond_rejoin_barrier_fires() { - // One source fans out through two unequal-latency producers (SMA(2), - // SMA(4)) that rejoin at a Barrier(0) node. Every push in a cycle carries - // that cycle's timestamp, so once both SMAs are warm and emit in the same - // cycle, both barrier inputs share the timestamp and the barrier fires — - // the within-source diamond rejoin the timestamp token handles (C6), - // distinct from the multi-source barrier above. - let build = || { + // One source fans out through SMA(2), SMA(4) that rejoin at a Barrier(0) + // node; node 3 = Recorder taps the barrier. Every push in a cycle carries + // that cycle's timestamp, so once both SMAs warm and emit in the same + // cycle, both barrier inputs share the timestamp and the barrier fires. + let build = |tx| { Harness::bootstrap( - vec![Box::new(Sma::new(2)), Box::new(Sma::new(4)), Box::new(BarrierSum { out: [Scalar::F64(0.0)] })], + vec![ + Box::new(Sma::new(2)), + Box::new(Sma::new(4)), + Box::new(BarrierSum { out: [Scalar::F64(0.0)] }), + Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)), + ], vec![SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }], }], - vec![Edge { from: 0, to: 2, slot: 0, from_field: 0 }, Edge { from: 1, to: 2, slot: 1, from_field: 0 }], - 2, + vec![ + Edge { from: 0, to: 2, slot: 0, from_field: 0 }, + Edge { from: 1, to: 2, slot: 1, from_field: 0 }, + Edge { from: 2, to: 3, slot: 0, from_field: 0 }, + ], ) .expect("valid DAG") }; let prices = f64_stream(&[(1, 10.0), (2, 12.0), (3, 14.0), (4, 16.0), (5, 18.0), (6, 20.0)]); - let mut h = build(); - let out = h.run(vec![prices.clone()]); + let (tx, rx) = mpsc::channel(); + let mut h = build(tx); + h.run(vec![prices.clone()]); + let out: Vec<(Timestamp, Vec)> = rx.try_iter().collect(); // SMA(4) warms at cycle 4; from then both paths emit each cycle at the same // timestamp, so the barrier fires: SMA(2)+SMA(4) = 15+13, 17+15, 19+17. assert_eq!( out, vec![ - None, - None, - None, - Some(vec![Scalar::F64(28.0)]), - Some(vec![Scalar::F64(32.0)]), - Some(vec![Scalar::F64(36.0)]), + (Timestamp(4), vec![Scalar::F64(28.0)]), + (Timestamp(5), vec![Scalar::F64(32.0)]), + (Timestamp(6), vec![Scalar::F64(36.0)]), ] ); - // determinism (C1): a second identical run is bit-identical - let mut h2 = build(); - assert_eq!(h2.run(vec![prices]), out); + let (tx2, rx2) = mpsc::channel(); + let mut h2 = build(tx2); + h2.run(vec![prices]); + let out2: Vec<(Timestamp, Vec)> = rx2.try_iter().collect(); + assert_eq!(out2, out); } #[test] fn mixed_a_and_b_or_combine_on_one_node() { - // in0,in1 = barrier group 0 (sources 0,1); in2 = as-of (source 2). + // MixedSum @0 (in0,in1 barrier group 0; in2 as-of); node 1 = Recorder taps it. + let (tx, rx) = mpsc::channel(); let mut h = Harness::bootstrap( - vec![Box::new(MixedSum { out: [Scalar::F64(0.0)] })], + vec![ + Box::new(MixedSum { out: [Scalar::F64(0.0)] }), + Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)), + ], vec![ SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }, SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 1 }] }, SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 2 }] }, ], - vec![], - 0, + vec![Edge { from: 0, to: 1, slot: 0, from_field: 0 }], ) .expect("valid"); let s0 = f64_stream(&[(2, 20.0), (5, 50.0)]); // in0 (barrier) let s1 = f64_stream(&[(2, 200.0)]); // in1 (barrier) let s2 = f64_stream(&[(1, 1.0), (3, 3.0)]); // in2 (as-of) - let out = h.run(vec![s0, s1, s2]); - // cycle order (ts, source idx): (1,s2) (2,s0) (2,s1) (3,s2) (5,s0). - // c3: barrier pair completes at t=2, holds c=1 -> 20+200+1 = 221. - // c4: as-of input ticks at t=3, holds the pair -> 20+200+3 = 223. - // c1 fires on the as-of input but filters (pair cold); c2,c5 hold. + h.run(vec![s0, s1, s2]); + let out: Vec<(Timestamp, Vec)> = rx.try_iter().collect(); + // c3: barrier pair completes at t=2, holds c=1 -> 221. c4: as-of input + // ticks at t=3, holds the pair -> 223. c1 filters; c2,c5 hold (no record). assert_eq!( out, vec![ - None, - None, - Some(vec![Scalar::F64(221.0)]), - Some(vec![Scalar::F64(223.0)]), - None, + (Timestamp(2), vec![Scalar::F64(221.0)]), + (Timestamp(3), vec![Scalar::F64(223.0)]), ] ); } @@ -718,7 +833,6 @@ mod tests { vec![Box::new(Sma::new(1)), Box::new(Sma::new(1))], vec![], vec![Edge { from: 0, to: 1, slot: 0, from_field: 0 }, Edge { from: 1, to: 0, slot: 0, from_field: 0 }], - 0, ) .unwrap_err(); assert_eq!(err, BootstrapError::Cycle); @@ -731,7 +845,6 @@ mod tests { vec![Box::new(Sma::new(1))], vec![SourceSpec { kind: ScalarKind::I64, targets: vec![Target { node: 0, slot: 0 }] }], vec![], - 0, ) .unwrap_err(); assert_eq!( @@ -742,12 +855,13 @@ mod tests { #[test] fn bootstrap_rejects_a_bad_index() { - // observe node 5 does not exist + // an edge target node (9) that does not exist -> BadIndex. (The old trigger + // — an out-of-range observe index — is gone with `observe`; BadIndex itself + // is unchanged, only the path that reaches it.) let err = Harness::bootstrap( - vec![Box::new(Sma::new(1))], + vec![Box::new(Sma::new(1)), Box::new(Sma::new(1))], vec![SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }], - vec![], - 5, + vec![Edge { from: 0, to: 9, slot: 0, from_field: 0 }], ) .unwrap_err(); assert_eq!(err, BootstrapError::BadIndex); @@ -775,36 +889,42 @@ mod tests { #[test] fn ohlcv_bundles_five_field_record() { - // node 0 = Ohlcv; five sources feed O/H/L/C/V; observe node 0. The barrier - // fires once all five share the timestamp, so each bar appears on the fifth - // cycle of its timestamp (the four partial cycles hold -> None). + // node 0 = Ohlcv; five sources feed O/H/L/C/V; node 1 = a 5-input Recorder + // taps all five fields via five edges. The barrier fires once all five share + // the timestamp, so each bar is recorded once, on the fifth cycle of its ts. + let (tx, rx) = mpsc::channel(); let mut h = Harness::bootstrap( - vec![Box::new(Ohlcv { out: [Scalar::F64(0.0); 5] })], + vec![ + Box::new(Ohlcv { out: [Scalar::F64(0.0); 5] }), + Box::new(Recorder::new( + &[ScalarKind::F64, ScalarKind::F64, ScalarKind::F64, ScalarKind::F64, ScalarKind::F64], + Firing::Any, + tx, + )), + ], ohlcv_sources(), - vec![], - 0, + vec![ + Edge { from: 0, to: 1, slot: 0, from_field: 0 }, // open + Edge { from: 0, to: 1, slot: 1, from_field: 1 }, // high + Edge { from: 0, to: 1, slot: 2, from_field: 2 }, // low + Edge { from: 0, to: 1, slot: 3, from_field: 3 }, // close + Edge { from: 0, to: 1, slot: 4, from_field: 4 }, // volume + ], ) .expect("valid"); - let out = h.run(ohlcv_streams()); + h.run(ohlcv_streams()); + let out: Vec<(Timestamp, Vec)> = rx.try_iter().collect(); assert_eq!( out, vec![ - None, - None, - None, - None, - Some(vec![ + (Timestamp(1), vec![ Scalar::F64(10.0), Scalar::F64(15.0), Scalar::F64(8.0), Scalar::F64(12.0), Scalar::F64(100.0), ]), - None, - None, - None, - None, - Some(vec![ + (Timestamp(2), vec![ Scalar::F64(20.0), Scalar::F64(25.0), Scalar::F64(19.0), @@ -817,82 +937,74 @@ mod tests { #[test] fn edge_binds_single_field_high_minus_low() { - // nodes [Ohlcv (0), Sub (1)]; Sub binds field 1 (high) and field 2 (low) - // of the Ohlcv record -> high - low == the bar range. Observing Sub proves - // from_field routes the right columns (not field 0), and that the two bound - // fields are co-fresh (Sub's Any inputs both fire in the bar's cycle). - let build = || { + // [Ohlcv (0), Sub (1), Recorder (2)]; Sub binds high (field 1) and low + // (field 2) of the Ohlcv record -> high - low; the Recorder taps Sub. + // Proves from_field routes the right columns (not field 0) and the two + // bound fields are co-fresh (Sub's Any inputs both fire in the bar's cycle). + let build = |tx| { Harness::bootstrap( vec![ Box::new(Ohlcv { out: [Scalar::F64(0.0); 5] }), Box::new(Sub::new()), + Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)), ], ohlcv_sources(), vec![ Edge { from: 0, to: 1, slot: 0, from_field: 1 }, // high Edge { from: 0, to: 1, slot: 1, from_field: 2 }, // low + Edge { from: 1, to: 2, slot: 0, from_field: 0 }, // Sub -> Recorder ], - 1, ) .expect("valid DAG") }; - let mut h = build(); - let out = h.run(ohlcv_streams()); + let (tx, rx) = mpsc::channel(); + let mut h = build(tx); + h.run(ohlcv_streams()); + let out: Vec<(Timestamp, Vec)> = rx.try_iter().collect(); // bar1: 15 - 8 = 7; bar2: 25 - 19 = 6 (each on the bar's fifth cycle). assert_eq!( out, vec![ - None, - None, - None, - None, - Some(vec![Scalar::F64(7.0)]), - None, - None, - None, - None, - Some(vec![Scalar::F64(6.0)]), + (Timestamp(1), vec![Scalar::F64(7.0)]), + (Timestamp(2), vec![Scalar::F64(6.0)]), ] ); - // determinism (C1): a second identical run is bit-identical - let mut h2 = build(); - assert_eq!(h2.run(ohlcv_streams()), out); + let (tx2, rx2) = mpsc::channel(); + let mut h2 = build(tx2); + h2.run(ohlcv_streams()); + let out2: Vec<(Timestamp, Vec)> = rx2.try_iter().collect(); + assert_eq!(out2, out); } #[test] fn distinct_edges_read_distinct_fields() { - // Same Ohlcv, a different consumer: Sub binds field 3 (close) and field 0 - // (open) -> close - open. Proves two different edges on one record read two - // different fields (3 and 0, neither of them the high/low pair above). + // Same Ohlcv, a different consumer: Sub binds close (field 3) and open + // (field 0) -> close - open; the Recorder taps Sub. Proves two edges on one + // record read two different fields (3 and 0, not the high/low pair above). + let (tx, rx) = mpsc::channel(); let mut h = Harness::bootstrap( vec![ Box::new(Ohlcv { out: [Scalar::F64(0.0); 5] }), Box::new(Sub::new()), + Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)), ], ohlcv_sources(), vec![ Edge { from: 0, to: 1, slot: 0, from_field: 3 }, // close Edge { from: 0, to: 1, slot: 1, from_field: 0 }, // open + Edge { from: 1, to: 2, slot: 0, from_field: 0 }, // Sub -> Recorder ], - 1, ) .expect("valid DAG"); - let out = h.run(ohlcv_streams()); + h.run(ohlcv_streams()); + let out: Vec<(Timestamp, Vec)> = rx.try_iter().collect(); // bar1: 12 - 10 = 2; bar2: 22 - 20 = 2. assert_eq!( out, vec![ - None, - None, - None, - None, - Some(vec![Scalar::F64(2.0)]), - None, - None, - None, - None, - Some(vec![Scalar::F64(2.0)]), + (Timestamp(1), vec![Scalar::F64(2.0)]), + (Timestamp(2), vec![Scalar::F64(2.0)]), ] ); } @@ -905,7 +1017,6 @@ mod tests { vec![Box::new(Sma::new(1)), Box::new(Sma::new(1))], vec![SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }], vec![Edge { from: 0, to: 1, slot: 0, from_field: 9 }], - 0, ) .unwrap_err(); assert_eq!(err, BootstrapError::BadIndex); @@ -920,7 +1031,284 @@ mod tests { vec![Box::new(TwoField { out: [Scalar::F64(0.0), Scalar::I64(0)] }), Box::new(Sma::new(1))], vec![SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }], vec![Edge { from: 0, to: 1, slot: 0, from_field: 1 }], - 1, + ) + .unwrap_err(); + assert_eq!( + err, + BootstrapError::KindMismatch { producer: ScalarKind::I64, consumer: ScalarKind::F64 } + ); + } + + #[test] + fn multi_sink_records_distinct_interior_streams() { + // Two recorders tap SMA(2) and SMA(4) in ONE run -> one run records many + // streams (the #2 headline). Each drained stream is individually correct. + let (tx_fast, rx_fast) = mpsc::channel(); + let (tx_slow, rx_slow) = mpsc::channel(); + let mut h = Harness::bootstrap( + vec![ + Box::new(Sma::new(2)), + Box::new(Sma::new(4)), + Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx_fast)), + Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx_slow)), + ], + vec![SourceSpec { + kind: ScalarKind::F64, + targets: vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }], + }], + vec![ + Edge { from: 0, to: 2, slot: 0, from_field: 0 }, // SMA(2) -> recorder fast + Edge { from: 1, to: 3, slot: 0, from_field: 0 }, // SMA(4) -> recorder slow + ], + ) + .expect("valid DAG"); + h.run(vec![f64_stream(&[(1, 10.0), (2, 12.0), (3, 14.0), (4, 16.0), (5, 18.0)])]); + let fast: Vec<(Timestamp, Vec)> = rx_fast.try_iter().collect(); + let slow: Vec<(Timestamp, Vec)> = rx_slow.try_iter().collect(); + // SMA(2) warms at cycle 2, SMA(4) at cycle 4 — two different-rate streams. + assert_eq!( + fast, + vec![ + (Timestamp(2), vec![Scalar::F64(11.0)]), + (Timestamp(3), vec![Scalar::F64(13.0)]), + (Timestamp(4), vec![Scalar::F64(15.0)]), + (Timestamp(5), vec![Scalar::F64(17.0)]), + ] + ); + assert_eq!( + slow, + vec![ + (Timestamp(4), vec![Scalar::F64(13.0)]), + (Timestamp(5), vec![Scalar::F64(15.0)]), + ] + ); + } + + #[test] + fn recorder_taps_all_fields_of_a_record() { + // A 5-input Recorder taps all five OHLCV fields via five field-wise edges + // (0005: N edges, no whole-record bind); its recorded row is the whole bar. + let (tx, rx) = mpsc::channel(); + let mut h = Harness::bootstrap( + vec![ + Box::new(Ohlcv { out: [Scalar::F64(0.0); 5] }), + Box::new(Recorder::new( + &[ScalarKind::F64, ScalarKind::F64, ScalarKind::F64, ScalarKind::F64, ScalarKind::F64], + Firing::Any, + tx, + )), + ], + ohlcv_sources(), + vec![ + Edge { from: 0, to: 1, slot: 0, from_field: 0 }, + Edge { from: 0, to: 1, slot: 1, from_field: 1 }, + Edge { from: 0, to: 1, slot: 2, from_field: 2 }, + Edge { from: 0, to: 1, slot: 3, from_field: 3 }, + Edge { from: 0, to: 1, slot: 4, from_field: 4 }, + ], + ) + .expect("valid"); + h.run(vec![ + f64_stream(&[(1, 10.0)]), + f64_stream(&[(1, 15.0)]), + f64_stream(&[(1, 8.0)]), + f64_stream(&[(1, 12.0)]), + f64_stream(&[(1, 100.0)]), + ]); + let out: Vec<(Timestamp, Vec)> = rx.try_iter().collect(); + assert_eq!(out.len(), 1); + assert_eq!(out[0].1.len(), 5); // all five fields recorded as one row + assert_eq!( + out, + vec![(Timestamp(1), vec![ + Scalar::F64(10.0), + Scalar::F64(15.0), + Scalar::F64(8.0), + Scalar::F64(12.0), + Scalar::F64(100.0), + ])] + ); + } + + #[test] + fn recorder_records_mixed_scalar_kinds() { + // A recorder with i64 + f64 + bool + timestamp inputs records a four-field + // mixed-kind row -> recording is not f64-only. Four sources tick once each + // at t=1,2,3,4; only on cycle 4 are all slots warm, so it records once, + // holding the earlier-ticked values. + let (tx, rx) = mpsc::channel(); + let mut h = Harness::bootstrap( + vec![Box::new(Recorder::new( + &[ScalarKind::I64, ScalarKind::F64, ScalarKind::Bool, ScalarKind::Timestamp], + Firing::Any, + tx, + ))], + vec![ + SourceSpec { kind: ScalarKind::I64, targets: vec![Target { node: 0, slot: 0 }] }, + SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 1 }] }, + SourceSpec { kind: ScalarKind::Bool, targets: vec![Target { node: 0, slot: 2 }] }, + SourceSpec { kind: ScalarKind::Timestamp, targets: vec![Target { node: 0, slot: 3 }] }, + ], + vec![], + ) + .expect("valid"); + h.run(vec![ + vec![(Timestamp(1), Scalar::I64(7))], + vec![(Timestamp(2), Scalar::F64(1.5))], + vec![(Timestamp(3), Scalar::Bool(true))], + vec![(Timestamp(4), Scalar::Ts(Timestamp(99)))], + ]); + let out: Vec<(Timestamp, Vec)> = rx.try_iter().collect(); + assert_eq!( + out, + vec![(Timestamp(4), vec![ + Scalar::I64(7), + Scalar::F64(1.5), + Scalar::Bool(true), + Scalar::Ts(Timestamp(99)), + ])] + ); + } + + #[test] + fn node_is_producer_and_sink_at_once() { + // TapForward records its input AND forwards it downstream; a second + // Recorder taps the forwarded output. Both channels see the same stream -> + // one node is producer and sink at once (C8 "both"). + let (tx_tap, rx_tap) = mpsc::channel(); + let (tx_down, rx_down) = mpsc::channel(); + let mut h = Harness::bootstrap( + vec![ + Box::new(TapForward { out: [Scalar::F64(0.0)], tx: tx_tap }), + Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx_down)), + ], + vec![SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }], + vec![Edge { from: 0, to: 1, slot: 0, from_field: 0 }], + ) + .expect("valid"); + h.run(vec![f64_stream(&[(1, 10.0), (2, 20.0), (3, 30.0)])]); + let tapped: Vec<(Timestamp, Vec)> = rx_tap.try_iter().collect(); + let downstream: Vec<(Timestamp, Vec)> = rx_down.try_iter().collect(); + let expected = vec![ + (Timestamp(1), vec![Scalar::F64(10.0)]), + (Timestamp(2), vec![Scalar::F64(20.0)]), + (Timestamp(3), vec![Scalar::F64(30.0)]), + ]; + assert_eq!(tapped, expected); // it recorded (sink side effect) + assert_eq!(downstream, expected); // and forwarded (producer output) + } + + #[test] + fn recording_is_deterministic() { + // Two fresh harnesses, two channels, identical input -> bit-identical + // recorded streams (C1). + let build = |tx| { + Harness::bootstrap( + vec![ + Box::new(Sma::new(3)), + Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)), + ], + vec![SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }], + vec![Edge { from: 0, to: 1, slot: 0, from_field: 0 }], + ) + .expect("valid") + }; + let prices = f64_stream(&[(1, 1.0), (2, 2.0), (3, 3.0), (4, 4.0), (5, 5.0)]); + + let (tx_a, rx_a) = mpsc::channel(); + let mut a = build(tx_a); + a.run(vec![prices.clone()]); + let run_a: Vec<(Timestamp, Vec)> = rx_a.try_iter().collect(); + + let (tx_b, rx_b) = mpsc::channel(); + let mut b = build(tx_b); + b.run(vec![prices]); + let run_b: Vec<(Timestamp, Vec)> = rx_b.try_iter().collect(); + + assert_eq!(run_a, run_b); + assert!(!run_a.is_empty()); // and it actually recorded something + } + + #[test] + fn recorder_barrier_firing_records_only_on_coincidence() { + // A 2-input Barrier(0) recorder records only on cycles where both inputs + // share the timestamp — the recorder's OWN firing policy gates recording. + let (tx, rx) = mpsc::channel(); + let mut h = Harness::bootstrap( + vec![Box::new(Recorder::new( + &[ScalarKind::F64, ScalarKind::F64], + Firing::Barrier(0), + tx, + ))], + vec![ + SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }, + SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 1 }] }, + ], + vec![], + ) + .expect("valid"); + let s0 = f64_stream(&[(1, 10.0), (2, 20.0), (3, 30.0), (4, 40.0)]); + let s1 = f64_stream(&[(2, 100.0), (4, 200.0)]); + h.run(vec![s0, s1]); + let out: Vec<(Timestamp, Vec)> = rx.try_iter().collect(); + // records ONLY at t=2 and t=4 (both inputs coincide); holds otherwise. + assert_eq!( + out, + vec![ + (Timestamp(2), vec![Scalar::F64(20.0), Scalar::F64(100.0)]), + (Timestamp(4), vec![Scalar::F64(40.0), Scalar::F64(200.0)]), + ] + ); + } + + #[test] + fn recorder_any_firing_records_on_each_fresh() { + // A 2-input Any recorder records on any-fresh once both are warm (as-of), + // holding the stale input. + let (tx, rx) = mpsc::channel(); + let mut h = Harness::bootstrap( + vec![Box::new(Recorder::new( + &[ScalarKind::F64, ScalarKind::F64], + Firing::Any, + tx, + ))], + vec![ + SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }, + SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 1 }] }, + ], + vec![], + ) + .expect("valid"); + let s0 = f64_stream(&[(1, 10.0), (2, 20.0), (3, 30.0), (4, 40.0)]); + let s1 = f64_stream(&[(2, 100.0), (4, 200.0)]); + h.run(vec![s0, s1]); + let out: Vec<(Timestamp, Vec)> = rx.try_iter().collect(); + // from t=2 on, records every cycle holding the stale input; two cycles fall + // on t=4 (the s0 tick then the s1 tick). + assert_eq!( + out, + vec![ + (Timestamp(2), vec![Scalar::F64(20.0), Scalar::F64(100.0)]), + (Timestamp(3), vec![Scalar::F64(30.0), Scalar::F64(100.0)]), + (Timestamp(4), vec![Scalar::F64(40.0), Scalar::F64(100.0)]), + (Timestamp(4), vec![Scalar::F64(40.0), Scalar::F64(200.0)]), + ] + ); + } + + #[test] + fn bootstrap_rejects_kind_mismatched_recorder_edge() { + // TwoField output: field 0 f64, field 1 i64. Binding field 1 (i64) into a + // Recorder's f64 input slot is a per-field kind mismatch -> KindMismatch + // (0005's check already covers recorder edges; recording adds no new hole). + let (tx, _rx) = mpsc::channel(); + let err = Harness::bootstrap( + vec![ + Box::new(TwoField { out: [Scalar::F64(0.0), Scalar::I64(0)] }), + Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)), + ], + vec![SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }], + vec![Edge { from: 0, to: 1, slot: 0, from_field: 1 }], ) .unwrap_err(); assert_eq!( diff --git a/crates/aura-std/src/sma.rs b/crates/aura-std/src/sma.rs index b7ed80f..caf3527 100644 --- a/crates/aura-std/src/sma.rs +++ b/crates/aura-std/src/sma.rs @@ -48,7 +48,7 @@ impl Node for Sma { #[cfg(test)] mod tests { use super::*; - use aura_core::AnyColumn; + use aura_core::{AnyColumn, Timestamp}; #[test] fn sma_warms_up_then_tracks_the_window_mean() { @@ -67,7 +67,7 @@ mod tests { for (v, want) in feed.iter().zip(expect) { inputs[0].push(Scalar::F64(*v)).unwrap(); - let got = sma.eval(Ctx::new(&inputs)); + 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())), @@ -81,9 +81,9 @@ 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)), Some([Scalar::F64(7.0)].as_slice())); + assert_eq!(sma.eval(Ctx::new(&inputs, Timestamp(0))), Some([Scalar::F64(7.0)].as_slice())); inputs[0].push(Scalar::F64(9.0)).unwrap(); - assert_eq!(sma.eval(Ctx::new(&inputs)), Some([Scalar::F64(9.0)].as_slice())); + assert_eq!(sma.eval(Ctx::new(&inputs, Timestamp(0))), Some([Scalar::F64(9.0)].as_slice())); } } diff --git a/crates/aura-std/src/sub.rs b/crates/aura-std/src/sub.rs index c013e58..5d161e0 100644 --- a/crates/aura-std/src/sub.rs +++ b/crates/aura-std/src/sub.rs @@ -49,7 +49,7 @@ impl Node for Sub { #[cfg(test)] mod tests { use super::*; - use aura_core::AnyColumn; + use aura_core::{AnyColumn, Timestamp}; #[test] fn sub_is_difference_once_both_inputs_present() { @@ -61,10 +61,10 @@ mod tests { // only input 0 present -> None inputs[0].push(Scalar::F64(10.0)).unwrap(); - assert_eq!(sub.eval(Ctx::new(&inputs)), None); + assert_eq!(sub.eval(Ctx::new(&inputs, Timestamp(0))), None); // both present -> a - b inputs[1].push(Scalar::F64(4.0)).unwrap(); - assert_eq!(sub.eval(Ctx::new(&inputs)), Some([Scalar::F64(6.0)].as_slice())); + assert_eq!(sub.eval(Ctx::new(&inputs, Timestamp(0))), Some([Scalar::F64(6.0)].as_slice())); } } diff --git a/docs/design/INDEX.md b/docs/design/INDEX.md index 41ea1f0..c74cec4 100644 --- a/docs/design/INDEX.md +++ b/docs/design/INDEX.md @@ -208,6 +208,16 @@ selects one producer column per edge; consuming a whole record is N edges (no 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). +**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 +the record to a destination it holds as a field (a channel, a chart handle) — an +**out-of-graph side effect**. There is no `Sink` type, trait, or engine flag: a +node that only records returns `None` (pure consumer), and a node may record +**and** return an output the engine forwards in the same `eval` (the "both" +case). In-graph routing stays engine-owned data (the edge table); the escape out +of the graph is the node's own side effect — and that boundary is the +determinism / graph-as-data boundary (C1/C7). ### C9 — Fractal, acyclic composition **Guarantee.** A composite is itself a `Node` that wires a sub-graph and exposes @@ -493,6 +503,15 @@ records. Making sinks the one recording-and-observability mechanism keeps "what can I see?" answerable by "what did I instrument?", and keeps the engine UI-agnostic (C14). Live param tuning (runtime values, no topology change — C12 / C19 Fork A) gives the interactive feel without a wiring DSL. +**Realization (cycle 0006).** Sinks-as-recording-mechanism is realized at the +substrate level: a recorded trace is exactly what a recording node pushed out of +the graph (no engine recording registry; the constructing World holds each +recording node's destination). The engine's single `observe: usize` affordance is +removed — `Harness::run` returns `()` and recording is a node-side concern, so one +run records *many* streams (one per recording node) instead of exactly one row. +Recorded streams are sparse and timestamped (a record per fired cycle, tagged +`ctx.now()`), matching a trace of timestamped events (C18). No new contract; the +`Harness` API change (observe removed, `run -> ()`) is recorded here. ---