diff --git a/crates/aura-engine/src/harness.rs b/crates/aura-engine/src/harness.rs index a40dbc1..eede29f 100644 --- a/crates/aura-engine/src/harness.rs +++ b/crates/aura-engine/src/harness.rs @@ -601,7 +601,7 @@ mod tests { // PortSpec / NodeSchema name the fixtures' declared signatures, brought in here // (production code reads them via the carried signatures, never naming the types). use aura_core::{FieldSpec, NodeSchema, PortSpec}; - use aura_std::{Bias, Recorder, Sma, SimBroker, Sub}; + use aura_std::{Bias, Recorder, Sma, SimBroker, Sub, When}; use std::sync::mpsc; /// Build an f64 source stream from (timestamp, value) points. @@ -876,6 +876,37 @@ mod tests { } } + /// Test gate: `f64 -> bool`, true iff the sample is strictly above `c`. + /// Derives a Bool gate stream **in-graph**, so the gate shares its + /// source's cycle (separate sources at the same stamp are separate + /// cycles — an in-graph derivation is how real gates arrive). + struct AboveGate { + c: f64, + out: [Cell; 1], + } + impl AboveGate { + fn sig() -> NodeSchema { + NodeSchema { + inputs: vec![f64_port(Firing::Any)], + output: vec![FieldSpec { name: "value".into(), kind: ScalarKind::Bool }], + params: vec![], + } + } + } + impl Node for AboveGate { + fn lookbacks(&self) -> Vec { + vec![1] + } + fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> { + let v = ctx.f64_in(0); + if v.is_empty() { + return None; + } + self.out[0] = Cell::from_bool(v[0] > self.c); + Some(&self.out) + } + } + /// Mixed A+B: barrier pair (inputs 0,1 in group 0) plus an as-of input /// (input 2). Fires when the pair completes (holding input 2) OR when input 2 /// ticks (holding the pair) — the OR-combine. @@ -1279,6 +1310,120 @@ mod tests { assert_eq!(out2, out); } + #[test] + fn when_gates_a_reducer_by_composition_equal_to_the_filtered_subsequence() { + // The composition claim the clock-enable design rests on (#281): + // SMA(When(x, gate), N) over a stream equals SMA(N) fed the gate-true + // subsequence directly. When's quiet cycles do not advance SMA's + // freshness, so the reducer state moves per forwarded sample, not per + // cycle — no gated variant of any reducer exists or is needed. + let prices = f64_stream(&[(1, 10.0), (2, 30.0), (3, 20.0), (4, 40.0), (5, 15.0), (6, 60.0)]); + // gate = price > 25.0 -> [f, t, f, t, f, t]; gated subsequence [30, 40, 60]. + let (tx, rx) = mpsc::channel(); + let mut h = boot( + vec![ + Box::new(AboveGate { c: 25.0, out: [Cell::from_bool(false)] }), + Box::new(When::new()), + Box::new(Sma::new(2)), + Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)), + ], + vec![ + AboveGate::sig(), + When::builder().schema().clone(), + Sma::builder().schema().clone(), + recorder_sig(&[ScalarKind::F64], Firing::Any), + ], + vec![SourceSpec::raw( + ScalarKind::F64, + vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }], + )], + vec![ + Edge { from: 0, to: 1, slot: 1, from_field: 0 }, + Edge { from: 1, to: 2, slot: 0, from_field: 0 }, + Edge { from: 2, to: 3, slot: 0, from_field: 0 }, + ], + ) + .expect("valid DAG"); + h.run(vec![Box::new(VecSource::new(prices))]); + let gated: Vec<(Timestamp, Vec)> = rx.try_iter().collect(); + + // SMA(2) over [30, 40, 60]: warms at the 2nd forwarded sample. + assert_eq!( + gated, + vec![ + (Timestamp(4), vec![Scalar::f64(35.0)]), + (Timestamp(6), vec![Scalar::f64(50.0)]), + ], + "gated SMA advances once per forwarded sample" + ); + + // Reference: the same SMA(2) fed the gated subsequence directly. + let filtered = f64_stream(&[(2, 30.0), (4, 40.0), (6, 60.0)]); + let (tx2, rx2) = mpsc::channel(); + let mut h2 = boot( + vec![ + Box::new(Sma::new(2)), + Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx2)), + ], + vec![Sma::builder().schema().clone(), recorder_sig(&[ScalarKind::F64], Firing::Any)], + vec![SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 0 }])], + vec![Edge { from: 0, to: 1, slot: 0, from_field: 0 }], + ) + .expect("valid DAG"); + h2.run(vec![Box::new(VecSource::new(filtered))]); + let reference: Vec<(Timestamp, Vec)> = rx2.try_iter().collect(); + + assert_eq!(gated, reference, "composition equals the filtered subsequence"); + } + + #[test] + fn a_quiet_when_stalls_its_barrier_group_for_the_cycle() { + // The deliberate stall pin (#281): a quiet When member is NOT at the + // current cycle timestamp, so its Barrier(0) group does not fire that + // cycle (the barrier contract); the group fires again on the next + // forwarded sample. + let prices = f64_stream(&[(1, 10.0), (2, 30.0), (3, 20.0), (4, 40.0)]); + // gate = price > 25.0 -> [f, t, f, t]; When forwards at t=2 and t=4 only. + let (tx, rx) = mpsc::channel(); + let mut h = boot( + vec![ + Box::new(AboveGate { c: 25.0, out: [Cell::from_bool(false)] }), + Box::new(When::new()), + Box::new(BarrierSum { out: [Cell::from_f64(0.0)] }), + Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)), + ], + vec![ + AboveGate::sig(), + When::builder().schema().clone(), + BarrierSum::sig(), + recorder_sig(&[ScalarKind::F64], Firing::Any), + ], + vec![SourceSpec::raw( + ScalarKind::F64, + vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }, Target { node: 2, slot: 1 }], + )], + vec![ + Edge { from: 0, to: 1, slot: 1, from_field: 0 }, + Edge { from: 1, to: 2, slot: 0, from_field: 0 }, + Edge { from: 2, to: 3, slot: 0, from_field: 0 }, + ], + ) + .expect("valid DAG"); + h.run(vec![Box::new(VecSource::new(prices))]); + let out: Vec<(Timestamp, Vec)> = rx.try_iter().collect(); + // t=1: gate false -> When quiet -> barrier member 0 never fresh -> no fire. + // t=2: When forwards 30; both members at t=2 -> fires 30+30=60. + // t=3: When quiet -> member 0 stays at t=2 -> the group STALLS (no t=3 row). + // t=4: When forwards 40; both members at t=4 -> fires 40+40=80. + assert_eq!( + out, + vec![ + (Timestamp(2), vec![Scalar::f64(60.0)]), + (Timestamp(4), vec![Scalar::f64(80.0)]), + ] + ); + } + #[test] fn mixed_a_and_b_or_combine_on_one_node() { // MixedSum @0 (in0,in1 barrier group 0; in2 as-of); node 1 = Recorder taps it.