feat(aura-std): Delay — C5 lag-N register

Emits the value its series input carried 'lag' fired cycles ago, held in a
node-owned ring (lookbacks()=[1], the past in node state not the input column).
Warm-up is skip-emit (None for the first 'lag' cycles — never a fabricated
sentinel). Output is a pure function of pre-cycle state, the precondition for a
future engine to close a feedback loop (the RTL register C5 names). prevHigh15 =
Delay[1] on high15. Build-step 4 of milestone 'Strategy node vocabulary I'.

closes #87

crates/aura-std/src/delay.rs

@@ -0,0 +1,173 @@
+//! `Delay` — the C5 register: outputs the value its single input had `lag` fired
+//! cycles ago (`prevHigh15` in the session-breakout strategy is `Delay[1]` on
+//! `high15`).
+//!
+//! The first **state** node. The lag history lives in a node-owned ring of length
+//! `lag` (a `Box<[f64]>` plus a `pos` write cursor and a `count` warm-up counter —
+//! the same shape and roles as `Sma`'s `ring`/`pos`/`count`), so `lookbacks()` is
+//! `1`: the input column is depth-1 and the past is held in node state, NOT read
+//! back out of the input column (that would be the input-column-lookback design,
+//! fork B; this is the node-owned-ring design, fork A).
+//!
+//! Why node-owned: the output at cycle T is then a pure function of state captured
+//! BEFORE T — the slot the next push will overwrite holds the lag-ago value, read
+//! out before the new sample lands. That precondition is what lets a future engine
+//! exclude a delay's in-edge from the topo sort and close a feedback loop (the
+//! RTL register C5 names): a delay's output never depends on the current cycle's
+//! upstream, only on history.
+//!
+//! Warm-up is **skip-emit**: `None` until `lag` samples have passed, because the
+//! lag-ago value does not exist yet — a `Delay[1]` has no previous bar on its
+//! first cycle. It NEVER fabricates a sentinel past value; a downstream comparator
+//! must not be handed a bar that never streamed.
+
+use aura_core::{
+    Cell, Ctx, FieldSpec, Firing, Node, NodeSchema, ParamSpec, PortSpec, PrimitiveBuilder,
+    ScalarKind,
+};
+
+/// The C5 lag-N register: emits the value its `series` input carried `lag` fired
+/// cycles ago, held in a node-owned ring. `None` during warm-up (the first `lag`
+/// cycles, when the lag-ago value does not yet exist).
+pub struct Delay {
+    lag: usize,
+    // The lag history, node-owned (so the input column is depth-1): `ring[pos]` is
+    // the value `lag` cycles ago — the one this cycle emits and the next push
+    // evicts. Sized once at construction (C7).
+    ring: Box<[f64]>,
+    pos: usize,
+    // Samples seen so far — the warm-up gate (silent until `lag`, like `Sma`).
+    count: usize,
+    out: [Cell; 1],
+}
+
+impl Delay {
+    /// Build a lag-`lag` register (must be >= 1: a zero-lag register is the
+    /// identity and is forbidden — mirror `Sma::new`'s `length >= 1`).
+    pub fn new(lag: usize) -> Self {
+        assert!(lag >= 1, "Delay lag must be >= 1");
+        Self {
+            lag,
+            ring: vec![0.0; lag].into_boxed_slice(),
+            pos: 0,
+            count: 0,
+            out: [Cell::from_f64(0.0)],
+        }
+    }
+
+    /// The param-generic recipe for a blueprint primitive: declares `lag` and
+    /// builds through `Delay::new` (the slice is kind-checked before `build` runs,
+    /// so the typed read is total).
+    pub fn builder() -> PrimitiveBuilder {
+        PrimitiveBuilder::new(
+            "Delay",
+            NodeSchema {
+                inputs: vec![PortSpec { kind: ScalarKind::F64, firing: Firing::Any, name: "series".into() }],
+                output: vec![FieldSpec { name: "value".into(), kind: ScalarKind::F64 }],
+                params: vec![ParamSpec { name: "lag".into(), kind: ScalarKind::I64 }],
+            },
+            |p| Box::new(Delay::new(p[0].i64() as usize)),
+        )
+    }
+}
+
+impl Node for Delay {
+    // The lag history lives in node state, so only the newest sample is read each
+    // cycle — `lag` sizes the ring, not the input column (like `Sma`).
+    fn lookbacks(&self) -> Vec<usize> {
+        vec![1]
+    }
+
+    fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> {
+        let w = ctx.f64_in(0);
+        if w.is_empty() {
+            return None; // empty-input cycle: no sample to register
+        }
+        let x = w[0]; // index 0 = newest (financial indexing)
+
+        // `ring[pos]` is the value `lag` cycles ago — read it OUT before the new
+        // sample overwrites it (so the output is a pure function of pre-cycle state,
+        // never the cycle's own upstream — the C5 feedback-loop precondition).
+        let lagged = self.ring[self.pos];
+        self.ring[self.pos] = x;
+        self.pos = (self.pos + 1) % self.lag;
+        if self.count < self.lag {
+            self.count += 1;
+            return None; // warm-up: the lag-ago value does not exist yet (skip-emit)
+        }
+
+        self.out[0] = Cell::from_f64(lagged);
+        Some(&self.out)
+    }
+
+    fn label(&self) -> String {
+        format!("Delay({})", self.lag)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use aura_core::{AnyColumn, Scalar, Timestamp};
+
+    #[test]
+    fn delay_one_emits_the_immediately_prior_value() {
+        // The headline property (this is `prevHigh15` = Delay[1] on high15): each
+        // fire emits the value the input carried the cycle before, and the first
+        // cycle is None — there is no previous bar yet (warm-up skip-emit, never a
+        // fabricated sentinel).
+        let node_for_depth = Delay::new(1);
+
+        // size the input column from the node's lookback, as bootstrap will at wiring
+        let mut inputs =
+            vec![AnyColumn::with_capacity(ScalarKind::F64, node_for_depth.lookbacks()[0])];
+        let mut node = node_for_depth;
+
+        let feed = [10.0_f64, 20.0, 30.0];
+        let expect = [None, Some(10.0), Some(20.0)];
+
+        for (v, want) in feed.iter().zip(expect) {
+            inputs[0].push(Scalar::f64(*v)).unwrap();
+            let got = node.eval(Ctx::new(&inputs, Timestamp(0)));
+            match want {
+                None => assert_eq!(got, None),
+                Some(p) => assert_eq!(got, Some([Cell::from_f64(p)].as_slice())),
+            }
+        }
+    }
+
+    #[test]
+    fn delay_two_emits_the_value_two_cycles_ago() {
+        // lag-N generalisation: Delay[2] holds two cycles of history, so it is None
+        // for the first two cycles, then emits the value from two cycles back.
+        let mut node = Delay::new(2);
+        let mut inputs = vec![AnyColumn::with_capacity(ScalarKind::F64, 1)];
+
+        let feed = [10.0_f64, 20.0, 30.0, 40.0];
+        let expect = [None, None, Some(10.0), Some(20.0)];
+
+        for (v, want) in feed.iter().zip(expect) {
+            inputs[0].push(Scalar::f64(*v)).unwrap();
+            let got = node.eval(Ctx::new(&inputs, Timestamp(0)));
+            match want {
+                None => assert_eq!(got, None),
+                Some(p) => assert_eq!(got, Some([Cell::from_f64(p)].as_slice())),
+            }
+        }
+    }
+
+    #[test]
+    fn delay_is_none_on_an_empty_input_cycle() {
+        // a degenerate empty-input cycle filters (the universal warm-up/empty guard)
+        let mut node = Delay::new(1);
+        let inputs = vec![AnyColumn::with_capacity(ScalarKind::F64, 1)];
+        assert_eq!(node.eval(Ctx::new(&inputs, Timestamp(0))), None);
+    }
+
+    #[test]
+    fn lag_must_be_at_least_one() {
+        // a zero-lag register is the identity and is forbidden (mirror Sma::new)
+        let r = std::panic::catch_unwind(|| Delay::new(0));
+        assert!(r.is_err(), "Delay::new(0) must panic — a zero-lag register is the identity");
+    }
+}

crates/aura-std/src/lib.rs

@@ -17,6 +17,7 @@
 
 mod add;
 mod and;
+mod delay;
 mod ema;
 mod eqconst;
 mod exposure;
@@ -28,6 +29,7 @@ mod sma;
 mod sub;
 pub use add::Add;
 pub use and::And;
+pub use delay::Delay;
 pub use ema::Ema;
 pub use eqconst::EqConst;
 pub use exposure::Exposure;