test(engine): pin When's gated-reducer and barrier-stall semantics

Two integration pins for the clock-enable design (#281), at the raw
FlatGraph boot seam. The gated-reducer equivalence is the crux claim:
SMA(When(x, gate), N) equals SMA(N) fed the gate-true subsequence
directly — When's quiet cycles do not advance the reducer's freshness,
so its state moves per forwarded sample, not per cycle. The barrier
stall pins that a quiet When member keeps its whole Barrier(0) group
from firing that cycle. An in-graph AboveGate fixture derives the Bool
gate co-fresh with the value leg.

refs #281

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<usize> {
+            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<Scalar>)> = 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<Scalar>)> = 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<Scalar>)> = 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.