feat(0070): streaming sink reductions — finalize hook + folding sinks (#138)

M3 machinery for BLOCKER #138: sink reductions can now fold online instead of
buffering every per-cycle row to an unbounded channel.

- aura-core: additive `Node::finalize` end-of-stream hook (default no-op) +
  `SeriesFold` online accumulator (last / max_drawdown / sign_flips).
- aura-engine: `Harness::run` calls `finalize` once per node in topo order after
  the source loop; `summarize` refactored to drive `SeriesFold` (byte-identical),
  the now-dead `sign0` removed.
- aura-std: `GatedRecorder` (emits only gated rows + the final row on finalize)
  and `SeriesReducer` (folds one f64 series, emits one summary row on finalize),
  both emitting the `Recorder` channel type — no aura-engine dependency.
- Integration tests: folded reductions are byte-for-byte equal to the
  raw-recorder path; the R-reduction's peak memory is O(trades), independent of
  cycle count.

Ratified two off-plan fixes the implementer made in the tests (both plan defects,
not code defects): the GatedRecorder correctly flushes the final non-gated row on
finalize, so the retained count is K_TRADES + 1; and the heterogeneous PM record
needs kind-correct zero scalars (a blanket f64 zero is rejected by the kind-typed
AnyColumn::push).

Wiring these reducers into the stage1-r sweep (the actual O(cycles)->O(trades)
win) follows next. refs #138

crates/aura-std/src/gated_recorder.rs

@@ -0,0 +1,146 @@
+//! `GatedRecorder` — a `Recorder` (C8 sink) that emits a recorded row only on
+//! cycles where a designated `bool` gate column is true, plus the genuine final
+//! row once on `finalize`. Fed a dense per-decision record gated on its
+//! "closed-this-cycle" column, it yields exactly the rows a trade-ledger
+//! reduction reads — the closed rows and the last row — in O(trades) memory
+//! instead of O(cycles). C7-pure: it holds only owned state + the channel.
+
+use aura_core::{Cell, Ctx, Firing, Node, NodeSchema, PortSpec, PrimitiveBuilder, Scalar, ScalarKind, Timestamp};
+use std::sync::mpsc::Sender;
+
+/// A recording sink that emits only rows whose `gate_col` (a `bool` column) is
+/// true, plus the final row once on `finalize` (deduped against the last gated
+/// emit). Warm-up behaves like `Recorder`: `None` until every column is warm,
+/// recording nothing.
+pub struct GatedRecorder {
+    kinds: Vec<ScalarKind>,
+    gate_col: usize,
+    last: Option<(Timestamp, Vec<Scalar>)>,
+    last_emitted: bool,
+    tx: Sender<(Timestamp, Vec<Scalar>)>,
+}
+
+impl GatedRecorder {
+    pub fn new(kinds: &[ScalarKind], gate_col: usize, tx: Sender<(Timestamp, Vec<Scalar>)>) -> Self {
+        Self { kinds: kinds.to_vec(), gate_col, last: None, last_emitted: false, tx }
+    }
+
+    /// Param-generic recipe. `kinds` / `gate_col` / `tx` are non-param construction
+    /// args (captured); the node declares no params. One input port per kind.
+    pub fn builder(
+        kinds: Vec<ScalarKind>,
+        gate_col: usize,
+        firing: Firing,
+        tx: Sender<(Timestamp, Vec<Scalar>)>,
+    ) -> PrimitiveBuilder {
+        let inputs = kinds
+            .iter()
+            .enumerate()
+            .map(|(i, &kind)| PortSpec { kind, firing, name: format!("col[{i}]") })
+            .collect();
+        let build_kinds = kinds.clone();
+        PrimitiveBuilder::new(
+            "GatedRecorder",
+            NodeSchema { inputs, output: vec![], params: vec![] }, // sink: empty output (C8)
+            move |_| Box::new(GatedRecorder::new(&build_kinds, gate_col, tx.clone())),
+        )
+    }
+}
+
+impl Node for GatedRecorder {
+    fn lookbacks(&self) -> Vec<usize> {
+        vec![1; self.kinds.len()]
+    }
+
+    fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> {
+        let mut row = Vec::with_capacity(self.kinds.len());
+        for (i, &kind) in self.kinds.iter().enumerate() {
+            let scalar = match kind {
+                ScalarKind::F64 => Scalar::f64(ctx.f64_in(i).get(0)?),
+                ScalarKind::I64 => Scalar::i64(ctx.i64_in(i).get(0)?),
+                ScalarKind::Bool => Scalar::bool(ctx.bool_in(i).get(0)?),
+                ScalarKind::Timestamp => Scalar::ts(ctx.ts_in(i).get(0)?),
+            };
+            row.push(scalar);
+        }
+        let gated = row[self.gate_col].as_bool();
+        if gated {
+            let _ = self.tx.send((ctx.now(), row.clone()));
+        }
+        self.last = Some((ctx.now(), row));
+        self.last_emitted = gated;
+        None
+    }
+
+    fn finalize(&mut self) {
+        // ensure a downstream `summarize_r` sees the true final row (e.g. an
+        // open-at-end position) without double-counting a final row already
+        // emitted as a closed trade.
+        if !self.last_emitted
+            && let Some(row) = self.last.take()
+        {
+            let _ = self.tx.send(row);
+        }
+    }
+
+    fn label(&self) -> String {
+        "GatedRecorder".to_string()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use aura_core::AnyColumn;
+    use std::sync::mpsc;
+
+    fn drive(seq: &[(bool, f64)]) -> Vec<(Timestamp, Vec<Scalar>)> {
+        let (tx, rx) = mpsc::channel();
+        let mut g = GatedRecorder::new(&[ScalarKind::Bool, ScalarKind::F64], 0, tx);
+        let mut inputs = vec![
+            AnyColumn::with_capacity(ScalarKind::Bool, 1),
+            AnyColumn::with_capacity(ScalarKind::F64, 1),
+        ];
+        for (t, &(gate, payload)) in seq.iter().enumerate() {
+            inputs[0].push(Scalar::bool(gate)).unwrap();
+            inputs[1].push(Scalar::f64(payload)).unwrap();
+            assert_eq!(g.eval(Ctx::new(&inputs, Timestamp(t as i64 + 1))), None);
+        }
+        g.finalize();
+        rx.try_iter().collect()
+    }
+
+    #[test]
+    fn emits_gated_rows_and_flushes_a_nongated_final_row() {
+        // hold, closed, hold(final). emits the closed row, then flushes the final hold.
+        let out = drive(&[(false, 1.0), (true, 2.0), (false, 3.0)]);
+        assert_eq!(
+            out,
+            vec![
+                (Timestamp(2), vec![Scalar::bool(true), Scalar::f64(2.0)]),
+                (Timestamp(3), vec![Scalar::bool(false), Scalar::f64(3.0)]),
+            ]
+        );
+    }
+
+    #[test]
+    fn does_not_double_emit_a_gated_final_row() {
+        // final row IS gated -> emitted once by the gate; finalize must not re-emit.
+        let out = drive(&[(false, 1.0), (true, 2.0)]);
+        assert_eq!(out, vec![(Timestamp(2), vec![Scalar::bool(true), Scalar::f64(2.0)])]);
+    }
+
+    #[test]
+    fn cold_warmup_records_nothing() {
+        let (tx, rx) = mpsc::channel();
+        let mut g = GatedRecorder::new(&[ScalarKind::Bool, ScalarKind::F64], 0, tx);
+        let inputs = vec![
+            AnyColumn::with_capacity(ScalarKind::Bool, 1),
+            AnyColumn::with_capacity(ScalarKind::F64, 1),
+        ];
+        // both columns cold -> None, nothing recorded, finalize flushes nothing.
+        assert_eq!(g.eval(Ctx::new(&inputs, Timestamp(1))), None);
+        g.finalize();
+        assert!(rx.try_recv().is_err());
+    }
+}

crates/aura-std/src/lib.rs

@@ -21,6 +21,7 @@ mod bias;
 mod delay;
 mod ema;
 mod eqconst;
+mod gated_recorder;
 mod gt;
 mod latch;
 mod lincomb;
@@ -29,6 +30,7 @@ mod mul;
 mod position_management;
 mod recorder;
 mod resample;
+mod series_reducer;
 mod session;
 mod sim_broker;
 mod sizer;
@@ -42,6 +44,7 @@ pub use bias::Bias;
 pub use delay::Delay;
 pub use ema::Ema;
 pub use eqconst::EqConst;
+pub use gated_recorder::GatedRecorder;
 pub use gt::Gt;
 pub use latch::Latch;
 pub use lincomb::LinComb;
@@ -53,6 +56,7 @@ pub use position_management::{
 };
 pub use recorder::Recorder;
 pub use resample::Resample;
+pub use series_reducer::SeriesReducer;
 pub use session::Session;
 pub use sim_broker::SimBroker;
 pub use sizer::Sizer;

crates/aura-std/src/series_reducer.rs

@@ -0,0 +1,94 @@
+//! `SeriesReducer` — a folding sink (C8) over one `f64` column. It folds each
+//! warm value into a `SeriesFold` and, on `finalize`, emits a single summary row
+//! `[last, max_drawdown, sign_flips]`, so a downstream reduction reads O(1) per
+//! member instead of the whole per-cycle series. C7-pure: owned fold + channel.
+
+use aura_core::{Cell, Ctx, Firing, Node, NodeSchema, PortSpec, PrimitiveBuilder, Scalar, ScalarKind, SeriesFold, Timestamp};
+use std::sync::mpsc::Sender;
+
+/// A single-`f64`-column folding sink. Emits exactly one row on `finalize`:
+/// `[Scalar::f64(last), Scalar::f64(max_drawdown), Scalar::i64(sign_flips)]`.
+pub struct SeriesReducer {
+    fold: SeriesFold,
+    last_ts: Timestamp,
+    tx: Sender<(Timestamp, Vec<Scalar>)>,
+}
+
+impl SeriesReducer {
+    pub fn new(tx: Sender<(Timestamp, Vec<Scalar>)>) -> Self {
+        Self { fold: SeriesFold::new(), last_ts: Timestamp(0), tx }
+    }
+
+    pub fn builder(firing: Firing, tx: Sender<(Timestamp, Vec<Scalar>)>) -> PrimitiveBuilder {
+        PrimitiveBuilder::new(
+            "SeriesReducer",
+            NodeSchema {
+                inputs: vec![PortSpec { kind: ScalarKind::F64, firing, name: "col[0]".to_string() }],
+                output: vec![], // sink: empty output (C8)
+                params: vec![],
+            },
+            move |_| Box::new(SeriesReducer::new(tx.clone())),
+        )
+    }
+}
+
+impl Node for SeriesReducer {
+    fn lookbacks(&self) -> Vec<usize> {
+        vec![1]
+    }
+
+    fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> {
+        let v = ctx.f64_in(0).get(0)?; // None during warm-up
+        self.fold.fold(v);
+        self.last_ts = ctx.now();
+        None
+    }
+
+    fn finalize(&mut self) {
+        let row = vec![
+            Scalar::f64(self.fold.last),
+            Scalar::f64(self.fold.max_drawdown),
+            Scalar::i64(self.fold.sign_flips as i64),
+        ];
+        let _ = self.tx.send((self.last_ts, row));
+    }
+
+    fn label(&self) -> String {
+        "SeriesReducer".to_string()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use aura_core::AnyColumn;
+    use std::sync::mpsc;
+
+    #[test]
+    fn folds_and_emits_one_summary_row_on_finalize() {
+        let (tx, rx) = mpsc::channel();
+        let mut r = SeriesReducer::new(tx);
+        let mut inputs = vec![AnyColumn::with_capacity(ScalarKind::F64, 1)];
+        for (t, v) in [(1_i64, 10.0_f64), (2, 12.0), (3, 8.0)] {
+            inputs[0].push(Scalar::f64(v)).unwrap();
+            assert_eq!(r.eval(Ctx::new(&inputs, Timestamp(t))), None);
+        }
+        // nothing emitted per-eval.
+        assert!(rx.try_recv().is_err());
+        r.finalize();
+        let out: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
+        assert_eq!(out.len(), 1);
+        assert_eq!(out[0].1[0], Scalar::f64(8.0)); // last
+        assert_eq!(out[0].1[1], Scalar::f64(4.0)); // max_drawdown 12 - 8
+        assert_eq!(out[0].1[2], Scalar::i64(0)); // sign_flips (all positive)
+    }
+
+    #[test]
+    fn empty_finalize_emits_zero_summary() {
+        let (tx, rx) = mpsc::channel();
+        let mut r = SeriesReducer::new(tx);
+        r.finalize();
+        let out: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
+        assert_eq!(out, vec![(Timestamp(0), vec![Scalar::f64(0.0), Scalar::f64(0.0), Scalar::i64(0)])]);
+    }
+}