feat(std): CumSum and When cells; roster complete at 33

Second half of the #281 clock-enable set. CumSum is the generic
run-length accumulator with O(1) Neumaier-compensated state — the
branching Neumaier variant, not Sma's plain Kahan, so it stays exact
when a large running total absorbs a tiny sample (the standalone
accumulator has no windowed-magnitude guarantee; module doc explains
why the helper is not reused). When is the clock-enable driver: forwards
value iff gate, else a quiet cycle (eval->None), so every stateful
reducer composes gated unmodified — SMA(When(x,g),N) is the gated SMA.

Rosters CumSum and When (both count pins 31->33) and ships two engine
E2E fixtures: When forwarding/quiet-downstream through the real
GraphBuilder->compile->run seam, and a quiet When leg suppressing a
whole Resample Barrier(0) group (the deliberate stall semantics).

refs #281

crates/aura-cli/tests/graph_construct.rs

@@ -121,8 +121,8 @@ fn graph_introspect_vocabulary_lists_exactly_the_closed_roster_count() {
     let lines: Vec<&str> = stdout.lines().filter(|l| !l.is_empty()).collect();
     assert_eq!(
         lines.len(),
-        31,
-        "the std-only (no project) vocabulary has exactly the roster's 31 entries: {stdout}"
+        33,
+        "the std-only (no project) vocabulary has exactly the roster's 33 entries: {stdout}"
     );
 }
 

crates/aura-engine/tests/cumsum_e2e.rs

@@ -0,0 +1,54 @@
+//! End-to-end coverage for the `CumSum` standard cell (#281): the generic
+//! running-total accumulator.
+//!
+//! The in-module tests in `aura-std/src/cumsum.rs` drive `eval` directly —
+//! they never go through a `PrimitiveBuilder` schema, a `GraphBuilder`-resolved
+//! wiring, or a compiled+bootstrapped `FlatGraph`. That is the real seam a
+//! data-authored op-script or a Rust builder actually exercises (C24); this
+//! file proves `CumSum` is reachable from the public `aura_engine`/`aura_std`
+//! construction surface, wires cleanly by declared port name/kind, and
+//! accumulates the right running total once actually driven cycle-by-cycle
+//! through a real `Harness::run` — the same bar the Sign/Select pair set in
+//! `select_sign_e2e.rs`.
+
+use std::sync::mpsc;
+
+use aura_core::{Firing, Scalar, ScalarKind, Timestamp};
+use aura_engine::{GraphBuilder, VecSource};
+use aura_std::{CumSum, Recorder};
+
+/// A deterministic tick stream, no timestamps/randomness beyond a fixed
+/// literal sequence.
+fn ticks(values: &[f64]) -> Vec<(Timestamp, Scalar)> {
+    values.iter().enumerate().map(|(i, &v)| (Timestamp(i as i64 + 1), Scalar::f64(v))).collect()
+}
+
+/// Property: **`CumSum` correctly registers its `PrimitiveBuilder` schema
+/// (one `f64` input, one `f64` output) and dispatches through a
+/// compiled+bootstrapped `FlatGraph`, accumulating the running total across
+/// real engine cycles** — not merely via a direct `eval` call (the in-module
+/// `cumsum.rs` unit tests' seam). A schema/wiring regression (wrong port
+/// kind, wrong output arity) would fail at `GraphBuilder::build` or
+/// `compile_with_params`, before a single cycle runs; a dispatch or
+/// state-carry regression would show up in the recorded running totals below.
+#[test]
+fn cumsum_dispatches_through_a_compiled_graph_and_accumulates() {
+    let (tx, rx) = mpsc::channel();
+    let mut g = GraphBuilder::new("root");
+    let series = g.source_role("series", ScalarKind::F64);
+    let cumsum = g.add(CumSum::builder());
+    let rec = g.add(Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx));
+
+    g.feed(series, [cumsum.input("series")]);
+    g.connect(cumsum.output("value"), rec.input("col[0]"));
+
+    let mut h = g.build().expect("resolves by name").bootstrap_with_params(vec![]).expect("bootstraps");
+    h.run(vec![Box::new(VecSource::new(ticks(&[2.0, -0.5, 3.5, 0.0])))]);
+    let trace: Vec<f64> = rx.try_iter().map(|(_, row)| row[0].as_f64()).collect();
+
+    assert_eq!(
+        trace,
+        vec![2.0, 1.5, 5.0, 5.0],
+        "running total accumulated across real engine cycles, dispatched through a compiled FlatGraph"
+    );
+}

crates/aura-engine/tests/when_e2e.rs

@@ -0,0 +1,133 @@
+//! End-to-end coverage for the `When` standard cell (#281): the clock-enable
+//! `f64 x bool -> f64` driver.
+//!
+//! The in-module tests in `aura-std/src/when.rs` drive `eval` directly — they
+//! never go through a `PrimitiveBuilder` schema, a `GraphBuilder`-resolved
+//! wiring, or a compiled+bootstrapped `FlatGraph`. That is the real seam a
+//! data-authored op-script or a Rust builder actually exercises (C24); this
+//! file proves `When` is reachable from the public `aura_engine`/`aura_std`
+//! construction surface, wires cleanly by declared port name/kind (`value:
+//! F64`, `gate: Bool`), and — the load-bearing property — that a quiet
+//! (gate-false) cycle never reaches a downstream sink: the recorder's own
+//! input only advances on a fresh push from `When`, so a quiet cycle leaves
+//! it un-fired rather than recording a held/zero value.
+//!
+//! One `price` source feeds both `When`'s `value` leg and, via
+//! `Gt(|price|, price)`, its `gate` leg (true exactly when `price < 0`) — the
+//! same single-source-per-cycle shape `select_sign_e2e.rs` uses, so `gate`
+//! and `value` are always computed in the same engine cycle (no cross-source
+//! phase skew between two independently-clocked inputs).
+use std::sync::mpsc;
+
+use aura_core::{Firing, Scalar, ScalarKind, Timestamp};
+use aura_engine::{GraphBuilder, VecSource};
+use aura_std::{Abs, Gt, Recorder, Resample, When};
+
+/// A deterministic `f64` tick stream, no timestamps/randomness beyond a fixed
+/// literal sequence.
+fn ticks(values: &[f64]) -> Vec<(Timestamp, Scalar)> {
+    values.iter().enumerate().map(|(i, &v)| (Timestamp(i as i64 + 1), Scalar::f64(v))).collect()
+}
+
+/// Property: **`When` correctly registers its `PrimitiveBuilder` schema
+/// (`value: F64`, `gate: Bool` in, one `f64` out) and dispatches through a
+/// compiled+bootstrapped `FlatGraph`, forwarding `value` on gate-true cycles
+/// and going quiet — not merely holding or zeroing — on gate-false cycles,
+/// all the way to a downstream sink.** A schema/wiring regression (wrong port
+/// kind, wrong output arity) would fail at `GraphBuilder::build` or
+/// `compile_with_params`, before a single cycle runs; a dispatch regression
+/// would show up in the recorded trace below — either the wrong values or,
+/// were quiet cycles wrongly forwarding a held/zero value, extra rows.
+#[test]
+fn when_dispatches_through_a_compiled_graph_and_is_quiet_downstream_on_gate_false() {
+    let (tx, rx) = mpsc::channel();
+    let mut g = GraphBuilder::new("root");
+    let price = g.source_role("price", ScalarKind::F64);
+    let abs = g.add(Abs::builder());
+    let gt = g.add(Gt::builder());
+    let when = g.add(When::builder());
+    let rec = g.add(Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx));
+
+    g.feed(price, [abs.input("value"), gt.input("b"), when.input("value")]);
+    g.connect(abs.output("value"), gt.input("a"));
+    g.connect(gt.output("value"), when.input("gate"));
+    g.connect(when.output("value"), rec.input("col[0]"));
+
+    let mut h = g.build().expect("resolves by name").bootstrap_with_params(vec![]).expect("bootstraps");
+    h.run(vec![Box::new(VecSource::new(ticks(&[-3.0, 2.5, -1.0, 4.0])))]);
+    let trace: Vec<f64> = rx.try_iter().map(|(_, row)| row[0].as_f64()).collect();
+
+    assert_eq!(
+        trace,
+        vec![-3.0, -1.0],
+        "only the two negative-price (gate-true) rows reach the sink; the positive-price rows are quiet, not recorded as held/zero"
+    );
+}
+
+/// Property: **a quiet `When` (gate false) driving one leg of a `Firing::Barrier(N)`
+/// group suppresses the WHOLE group that cycle — not merely its own leg.** This is
+/// the composition the `when.rs` module doc calls out directly ("a quiet `When`
+/// feeding a `Firing::Barrier(N)` group means the group does not fire that cycle").
+/// `Resample`'s four OHLC inputs are `aura-std`'s only `Barrier(0)` consumer; here
+/// `When` drives `close` while `open`/`high`/`low` are wired straight from the same
+/// source (fresh every cycle, gate or no gate). On a gate-false tick the other three
+/// legs still receive a fresh push, but `close` does not — since `Barrier(0)`
+/// requires all four co-fresh, the node does not fire at all that cycle, so the
+/// gate-false tick's price never enters the OHLC accumulator (not even via the
+/// ungated legs), and a bucket rollover due exactly on a gate-false tick is
+/// deferred to the next gate-true tick rather than firing early or partially.
+#[test]
+fn quiet_when_leg_suppresses_the_whole_barrier_group() {
+    let (tx, rx) = mpsc::channel();
+    let mut g = GraphBuilder::new("root");
+    let price = g.source_role("price", ScalarKind::F64);
+    let abs = g.add(Abs::builder());
+    let gt = g.add(Gt::builder());
+    let when = g.add(When::builder());
+    let resample = g.add(Resample::builder());
+    let rec = g.add(Recorder::builder(vec![ScalarKind::F64; 4], Firing::Any, tx));
+
+    g.feed(price, [
+        abs.input("value"),
+        gt.input("b"),
+        when.input("value"),
+        resample.input("open"),
+        resample.input("high"),
+        resample.input("low"),
+    ]);
+    g.connect(abs.output("value"), gt.input("a"));
+    g.connect(gt.output("value"), when.input("gate"));
+    g.connect(when.output("value"), resample.input("close"));
+    g.connect(resample.output("open"), rec.input("col[0]"));
+    g.connect(resample.output("high"), rec.input("col[1]"));
+    g.connect(resample.output("low"), rec.input("col[2]"));
+    g.connect(resample.output("close"), rec.input("col[3]"));
+
+    let mut h = g
+        .build()
+        .expect("resolves by name")
+        .bootstrap_with_params(vec![Scalar::i64(1)]) // Resample's sole open param: period_minutes=1
+        .expect("bootstraps");
+
+    // gate = price < 0 (Gt(|price|, price)); 1-minute buckets (period_ns = 60e9).
+    let feed: [(i64, f64); 5] = [
+        (0, -1.0),               // gate true: bucket 0 opens (o=h=l=c=-1.0)
+        (10_000_000_000, 2.0),   // gate false: When quiet -> Resample does not fire at all
+        (20_000_000_000, -3.0),  // gate true, still bucket 0: high stays -1.0, low->-3.0, close->-3.0
+        (60_000_000_000, 4.0),   // gate false exactly on the bucket-1 boundary: rollover DEFERRED
+        (70_000_000_000, -6.0),  // gate true, bucket 1: fires -> rolls over the completed bucket-0 bar
+    ];
+    let ticks: Vec<(Timestamp, Scalar)> =
+        feed.iter().map(|&(t, v)| (Timestamp(t), Scalar::f64(v))).collect();
+    h.run(vec![Box::new(VecSource::new(ticks))]);
+    let trace: Vec<Vec<f64>> =
+        rx.try_iter().map(|(_, row)| row.iter().map(|c| c.as_f64()).collect()).collect();
+
+    assert_eq!(
+        trace,
+        vec![vec![-1.0, -1.0, -3.0, -3.0]],
+        "the two gate-false ticks (price 2.0, 4.0) must never enter the OHLC accumulator: \
+         a quiet When leg blocks the whole Barrier(0) group, not just its own field, and the \
+         bucket-1 rollover due at t=60e9 is deferred to the next gate-true tick (t=70e9)"
+    );
+}

crates/aura-std/src/cumsum.rs

@@ -0,0 +1,145 @@
+//! `CumSum` — running total of one `f64` input with O(1) scalar state.
+//!
+//! The generic run-length accumulator (#281). The domain-embedded precedents
+//! (`SimBroker.cum`, `PositionManagement.cum_realized_r`, `CostRunner.cum`)
+//! stay as they are; this cell is the vocabulary form, so counters compose:
+//! `count_true = CumSum(Select(cond, 1.0, 0.0))`.
+//!
+//! The running sum carries a compensation term for the same reason
+//! [`Sma`](crate::Sma) documents: an accumulative running sum has nowhere for
+//! rounding error to go, so over millions of adds an uncompensated total
+//! drifts. The compensated total is folded in at the read (`sum + comp`); a
+//! `NaN` input poisons the total (same posture as `Sma`, not guarded).
+//!
+//! **Neumaier, not Sma's plain Kahan.** `Sma`'s private `kahan()` helper uses
+//! the branchless Kahan formula, sound because a windowed mean's addends stay
+//! the same order of magnitude as the running window sum. A free-standing
+//! accumulator has no such guarantee — a large running total can absorb a
+//! comparatively tiny sample (`1e16 + 1.0`), or the reverse — so this cell uses
+//! the branching Neumaier variant, which stays exact in *both* magnitude orders
+//! (`|sum| >= |x|` and `|x| > |sum|`). Plain Kahan drops the small addend in the
+//! latter case (`[1e16, 1.0, -1e16]` -> `0.0`, not `1.0`); Neumaier keeps it.
+//! That is why the helper is not reused here — the algorithms differ, and this
+//! one is deliberately the more robust of the two.
+//!
+//! One `f64` input (`Firing::Any`), one `f64` output, no params. Emits from
+//! the first sample (`cum = x0`); state advances once per fired eval (the
+//! engine's freshness gating guarantees one eval per fresh sample).
+
+use aura_core::{Cell, Ctx, FieldSpec, Firing, Node, NodeSchema, PortSpec, PrimitiveBuilder, ScalarKind};
+
+/// Neumaier-compensated running total: emits `x0 + x1 + ... + xn` at each
+/// fired cycle. Emits `None` until the input is warm (C8).
+pub struct CumSum {
+    // Running sum and its Neumaier compensation term (the low-order bits each
+    // add dropped, folded into the emitted total at the read).
+    sum: f64,
+    comp: f64,
+    out: [Cell; 1],
+}
+
+impl CumSum {
+    /// Build a `CumSum` node, initially zero.
+    pub fn new() -> Self {
+        Self { sum: 0.0, comp: 0.0, out: [Cell::from_f64(0.0)] }
+    }
+
+    /// The param-generic recipe for a blueprint primitive: paramless, builds
+    /// through `CumSum::new`.
+    pub fn builder() -> PrimitiveBuilder {
+        PrimitiveBuilder::new(
+            "CumSum",
+            NodeSchema {
+                inputs: vec![PortSpec { kind: ScalarKind::F64, firing: Firing::Any, name: "series".into() }],
+                output: vec![FieldSpec { name: "value".into(), kind: ScalarKind::F64 }],
+                params: vec![],
+            },
+            |_| Box::new(CumSum::new()),
+        )
+    }
+}
+
+impl Default for CumSum {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl Node for CumSum {
+    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; // warm-up gate (C8)
+        }
+        let x = v[0];
+        // Neumaier add: capture the low-order bits the naive add drops.
+        let t = self.sum + x;
+        if self.sum.abs() >= x.abs() {
+            self.comp += (self.sum - t) + x;
+        } else {
+            self.comp += (x - t) + self.sum;
+        }
+        self.sum = t;
+        self.out[0] = Cell::from_f64(self.sum + self.comp);
+        Some(&self.out)
+    }
+
+    fn label(&self) -> String {
+        "CumSum".to_string()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use aura_core::{AnyColumn, Scalar, Timestamp};
+
+    #[test]
+    fn cumsum_accumulates_from_the_first_sample() {
+        // PROPERTY: emits the running total from the very first sample on,
+        // advancing once per fired eval. Exact-representable values keep the
+        // compensation term at zero, so the expectations are exact.
+        let mut node = CumSum::new();
+        let mut inputs = vec![AnyColumn::with_capacity(ScalarKind::F64, 1)];
+        let feed = [2.0, -0.5, 3.5, 0.0];
+        let want = [2.0, 1.5, 5.0, 5.0];
+        for (x, w) in feed.iter().zip(want) {
+            inputs[0].push(Scalar::f64(*x)).unwrap();
+            let got = node.eval(Ctx::new(&inputs, Timestamp(0)));
+            assert_eq!(got, Some([Cell::from_f64(w)].as_slice()));
+        }
+    }
+
+    #[test]
+    fn cumsum_is_none_until_warm() {
+        let mut node = CumSum::new();
+        let inputs = vec![AnyColumn::with_capacity(ScalarKind::F64, 1)];
+        assert_eq!(node.eval(Ctx::new(&inputs, Timestamp(0))), None);
+    }
+
+    #[test]
+    fn cumsum_compensation_survives_magnitude_spread() {
+        // The Neumaier pin: 1e16 + 1.0 == 1e16 in naive f64 (the 1.0 is below
+        // the ULP), so a naive running sum over [1e16, 1.0, -1e16] ends at 0.0.
+        // The compensated total keeps the small addend and ends at exactly 1.0.
+        let mut node = CumSum::new();
+        let mut inputs = vec![AnyColumn::with_capacity(ScalarKind::F64, 1)];
+        let mut last = 0.0;
+        for x in [1e16, 1.0, -1e16] {
+            inputs[0].push(Scalar::f64(x)).unwrap();
+            last = node.eval(Ctx::new(&inputs, Timestamp(0))).unwrap()[0].f64();
+        }
+        assert_eq!(last, 1.0, "the compensated running sum keeps the small addend");
+    }
+
+    #[test]
+    fn input_slot_is_named_series() {
+        let names: Vec<String> =
+            CumSum::builder().schema().inputs.iter().map(|p| p.name.clone()).collect();
+        assert_eq!(names, ["series"]);
+    }
+}

crates/aura-std/src/lib.rs

@@ -26,6 +26,7 @@ mod const_node;
 mod constant_cost;
 mod cost;
 mod cost_sum;
+mod cumsum;
 mod delay;
 mod div;
 mod ema;
@@ -57,6 +58,7 @@ mod sub;
 mod vocabulary;
 mod vol_slippage_cost;
 mod vol_tf_stop;
+mod when;
 pub use abs::Abs;
 pub use add::Add;
 pub use and::And;
@@ -69,6 +71,7 @@ pub use cost::{
     COST_FIELD_NAMES, COST_WIDTH, GEOMETRY_WIDTH,
 };
 pub use cost_sum::CostSum;
+pub use cumsum::CumSum;
 pub use delay::Delay;
 pub use div::Div;
 pub use ema::Ema;
@@ -103,3 +106,4 @@ pub use sub::Sub;
 pub use vocabulary::{std_vocabulary, std_vocabulary_types};
 pub use vol_slippage_cost::VolSlippageCost;
 pub use vol_tf_stop::VolTfStop;
+pub use when::When;

crates/aura-std/src/vocabulary.rs

@@ -18,9 +18,9 @@
 //! later, additive extension (#156/C20).
 
 use crate::{
-    Abs, Add, And, Bias, CarryCost, Const, ConstantCost, Delay, Div, Ema, EqConst, FixedStop, Gt,
-    Latch, LongOnly, Max, Min, Mul, PositionManagement, Resample, RollingMax, RollingMin, Scale,
-    Select, SessionFrankfurt, Sign, Sizer, Sma, Sqrt, Sub, VolSlippageCost,
+    Abs, Add, And, Bias, CarryCost, Const, ConstantCost, CumSum, Delay, Div, Ema, EqConst,
+    FixedStop, Gt, Latch, LongOnly, Max, Min, Mul, PositionManagement, Resample, RollingMax,
+    RollingMin, Scale, Select, SessionFrankfurt, Sign, Sizer, Sma, Sqrt, Sub, VolSlippageCost, When,
 };
 use aura_core::PrimitiveBuilder;
 
@@ -65,6 +65,7 @@ std_vocabulary_roster! {
     "CarryCost" => CarryCost,
     "Const" => Const,
     "ConstantCost" => ConstantCost,
+    "CumSum" => CumSum,
     "Delay" => Delay,
     "Div" => Div,
     "EMA" => Ema,
@@ -89,6 +90,7 @@ std_vocabulary_roster! {
     "Sqrt" => Sqrt,
     "Sub" => Sub,
     "VolSlippageCost" => VolSlippageCost,
+    "When" => When,
 }
 
 #[cfg(test)]
@@ -132,7 +134,7 @@ mod tests {
         assert!(!std_vocabulary_types().contains(&"LinComb")); // construction-arg node
         assert!(!std_vocabulary_types().contains(&"Recorder")); // sink
         assert!(!std_vocabulary_types().contains(&"nope"));
-        // count guard: pins the roster at exactly 31 entries
-        assert_eq!(std_vocabulary_types().len(), 31);
+        // count guard: pins the roster at exactly 33 entries
+        assert_eq!(std_vocabulary_types().len(), 33);
     }
 }

crates/aura-std/src/when.rs

@@ -0,0 +1,139 @@
+//! `When` — the clock-enable driver `f64 × bool -> f64`: forwards `value`
+//! verbatim when `gate` is true, and is **quiet** (`eval -> None`, the
+//! `Resample` precedent) when it is false.
+//!
+//! The single clock-enable standard cell of the RTL metaphor (#281): gating
+//! belongs in the *stream*, not in per-node gated variants. Because the run
+//! loop is freshness-gated, every existing stateful reducer composes gated,
+//! unmodified: `SMA(When(x, gate), N)` IS the gated SMA — under
+//! `Firing::Any` a consumer holds its last forwarded value across quiet
+//! cycles and its state advances per *firing*, not per cycle. A quiet
+//! `When` feeding a `Firing::Barrier(N)` group means the group does not
+//! fire that cycle — the quiet member is not at the current timestamp,
+//! which is the correct reading of the barrier contract (deliberately
+//! pinned by an engine test).
+//!
+//! Two inputs (`value: F64`, `gate: Bool`, each `Firing::Any` — the enable
+//! is level-sensitive: the gate *level* is sampled whenever the cell fires,
+//! however long ago it was driven). One `f64` output, no params, stateless.
+//! Emits `None` while either leg is cold (warm-up), and on every gate-false
+//! cycle (the quiet cycle that IS the semantics).
+
+use aura_core::{Cell, Ctx, FieldSpec, Firing, Node, NodeSchema, PortSpec, PrimitiveBuilder, ScalarKind};
+
+/// Clock-enable forwarder: `value` iff `gate`, else quiet (`None`). A false
+/// gate is a quiet cycle, not a held or zero output — downstream freshness
+/// does not advance.
+pub struct When {
+    out: [Cell; 1],
+}
+
+impl When {
+    /// Build a `When` node.
+    pub fn new() -> Self {
+        Self { out: [Cell::from_f64(0.0)] }
+    }
+
+    /// The param-generic recipe for a blueprint primitive: paramless, builds
+    /// through `When::new`.
+    pub fn builder() -> PrimitiveBuilder {
+        PrimitiveBuilder::new(
+            "When",
+            NodeSchema {
+                inputs: vec![
+                    PortSpec { kind: ScalarKind::F64, firing: Firing::Any, name: "value".into() },
+                    PortSpec { kind: ScalarKind::Bool, firing: Firing::Any, name: "gate".into() },
+                ],
+                output: vec![FieldSpec { name: "value".into(), kind: ScalarKind::F64 }],
+                params: vec![],
+            },
+            |_| Box::new(When::new()),
+        )
+    }
+}
+
+impl Default for When {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl Node for When {
+    fn lookbacks(&self) -> Vec<usize> {
+        vec![1, 1]
+    }
+
+    fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> {
+        let value = ctx.f64_in(0);
+        let gate = ctx.bool_in(1);
+        if value.is_empty() || gate.is_empty() {
+            return None; // cold warm-up (C8) — both legs required
+        }
+        if !gate[0] {
+            return None; // gate low -> QUIET cycle (the semantics, not warm-up)
+        }
+        self.out[0] = Cell::from_f64(value[0]);
+        Some(&self.out)
+    }
+
+    fn label(&self) -> String {
+        "When".to_string()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use aura_core::{AnyColumn, Scalar, Timestamp};
+
+    fn gate_inputs() -> Vec<AnyColumn> {
+        vec![
+            AnyColumn::with_capacity(ScalarKind::F64, 1),
+            AnyColumn::with_capacity(ScalarKind::Bool, 1),
+        ]
+    }
+
+    #[test]
+    fn when_forwards_on_true_and_is_quiet_on_false() {
+        // PROPERTY: gate true -> the value passes through verbatim (bit-equal);
+        // gate false -> a QUIET cycle (None), never a held or zero output.
+        let mut node = When::new();
+        let mut inputs = gate_inputs();
+
+        inputs[0].push(Scalar::f64(42.5)).unwrap();
+        inputs[1].push(Scalar::bool(true)).unwrap();
+        assert_eq!(
+            node.eval(Ctx::new(&inputs, Timestamp(0))),
+            Some([Cell::from_f64(42.5)].as_slice())
+        );
+
+        inputs[0].push(Scalar::f64(7.0)).unwrap();
+        inputs[1].push(Scalar::bool(false)).unwrap();
+        assert_eq!(node.eval(Ctx::new(&inputs, Timestamp(0))), None);
+
+        inputs[0].push(Scalar::f64(-3.25)).unwrap();
+        inputs[1].push(Scalar::bool(true)).unwrap();
+        assert_eq!(
+            node.eval(Ctx::new(&inputs, Timestamp(0))),
+            Some([Cell::from_f64(-3.25)].as_slice())
+        );
+    }
+
+    #[test]
+    fn when_is_none_while_either_leg_is_cold() {
+        // Cold warm-up is distinct from the quiet cycle: before both legs have
+        // seen a value the cell is cold (C8), regardless of the gate.
+        let mut node = When::new();
+        let mut inputs = gate_inputs();
+        assert_eq!(node.eval(Ctx::new(&inputs, Timestamp(0))), None); // both cold
+        inputs[0].push(Scalar::f64(1.0)).unwrap();
+        assert_eq!(node.eval(Ctx::new(&inputs, Timestamp(0))), None); // gate cold
+    }
+
+    #[test]
+    fn input_slots_are_named_value_gate() {
+        let names: Vec<String> =
+            When::builder().schema().inputs.iter().map(|p| p.name.clone()).collect();
+        assert_eq!(names, ["value", "gate"]);
+    }
+}