diff --git a/docs/plans/0070-streaming-sink-reductions.md b/docs/plans/0070-streaming-sink-reductions.md new file mode 100644 index 0000000..2854caa --- /dev/null +++ b/docs/plans/0070-streaming-sink-reductions.md @@ -0,0 +1,1065 @@ +# Streaming Sink Reductions (BLOCKER #138) — Implementation Plan + +> **Parent spec:** `docs/specs/0070-streaming-sink-reductions.md` +> +> **For agentic workers:** REQUIRED SUB-SKILL: use the `implement` skill to run +> this plan. Steps use `- [ ]` checkboxes for tracking. + +**Goal:** Make a full-history no-trace `stage1-r` sweep run in O(trades)+O(1) +memory per member by folding sink reductions online behind a new `finalize` +end-of-stream lifecycle, with every existing metric value byte-identical. + +**Architecture:** A default-no-op `Node::finalize` hook (aura-core) called once +per node by `Harness::run` after the source loop drains; a shared `SeriesFold` +online accumulator (aura-core) that both `summarize` and a new `SeriesReducer` +sink drive; a `GatedRecorder` sink that emits only gated (closed) rows + the +final row, fed verbatim to the unchanged `summarize_r`. The no-trace sweep wires +these reducers; the `--trace` path and single run keep raw recorders as the +byte-for-byte reference. + +**Tech Stack:** aura-core (trait + fold), aura-engine (harness epilogue, +summarize refactor, integration tests), aura-std (two sink nodes), aura-cli +(graph rewiring). + +**Files this plan creates or modifies:** + +- Create: `crates/aura-core/src/series_fold.rs` — `SeriesFold` online accumulator + `sign0` + unit tests +- Modify: `crates/aura-core/src/lib.rs:31-47` — `mod series_fold;` + `pub use series_fold::SeriesFold;` +- Modify: `crates/aura-core/src/node.rs:280-283` — add `fn finalize(&mut self) {}` default + a unit test +- Modify: `crates/aura-engine/src/harness.rs:443` — post-loop `finalize()` epilogue + a `FinalizeProbe` test +- Modify: `crates/aura-engine/src/report.rs:369-416` — `summarize` drives `SeriesFold`; remove dead `sign0` +- Create: `crates/aura-std/src/gated_recorder.rs` — `GatedRecorder` sink + unit tests +- Create: `crates/aura-std/src/series_reducer.rs` — `SeriesReducer` sink + unit tests +- Modify: `crates/aura-std/src/lib.rs:18-62` — `mod`+`pub use` for the two sinks +- Create: `crates/aura-engine/tests/streaming_reduction_equivalence.rs` — folded == raw, field-for-field +- Create: `crates/aura-engine/tests/streaming_reduction_memory.rs` — counting-allocator O(trades) capstone +- Modify: `crates/aura-cli/src/main.rs:1924-1988` — `stage1_r_graph` gains `reduce: bool` +- Modify: `crates/aura-cli/src/main.rs:1179,1206-1242,2006` — thread `reduce`; folded no-trace consumer + +--- + +### Task 1: `SeriesFold` online accumulator (aura-core) + +**Files:** +- Create: `crates/aura-core/src/series_fold.rs` +- Modify: `crates/aura-core/src/lib.rs:31-47` + +- [ ] **Step 1: Write the new module with impl + failing tests** + +Create `crates/aura-core/src/series_fold.rs`: + +```rust +//! `SeriesFold` — the online accumulator behind `summarize`'s three pip +//! statistics (`last`, `max_drawdown`, `sign_flips`). One arithmetic source of +//! truth, foldable a value at a time, so a slice driver (`summarize`) and a +//! per-`eval` streaming sink (`SeriesReducer`) compute byte-identical results. + +/// Three-way sign: `-1.0` / `0.0` / `+1.0`. A zero maps to `0.0` so flat is +/// distinct from long/short in the flip count (unlike `f64::signum`). +fn sign0(v: f64) -> f64 { + if v > 0.0 { + 1.0 + } else if v < 0.0 { + -1.0 + } else { + 0.0 + } +} + +/// Online fold of a single `f64` series into the pip summary statistics. The +/// operations and their order mirror `summarize`'s loops exactly, so a +/// slice-driven and a per-value-driven fold agree bit-for-bit (IEEE-754 +/// non-associativity respected). +#[derive(Clone, Debug)] +pub struct SeriesFold { + /// the last value folded (`total_pips` for an equity series); `0.0` if none. + pub last: f64, + peak: f64, + /// running `max(peak - value)`, always `>= 0.0`. + pub max_drawdown: f64, + prev_sign: Option, + /// count of adjacent normalized-sign changes. + pub sign_flips: u64, +} + +impl Default for SeriesFold { + fn default() -> Self { + Self { last: 0.0, peak: f64::NEG_INFINITY, max_drawdown: 0.0, prev_sign: None, sign_flips: 0 } + } +} + +impl SeriesFold { + pub fn new() -> Self { + Self::default() + } + + /// Fold one value into the running statistics. + pub fn fold(&mut self, v: f64) { + self.last = v; + if v > self.peak { + self.peak = v; + } + let dd = self.peak - v; + if dd > self.max_drawdown { + self.max_drawdown = dd; + } + let s = sign0(v); + if let Some(p) = self.prev_sign + && s != p + { + self.sign_flips += 1; + } + self.prev_sign = Some(s); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn series_fold_empty_is_zero() { + let f = SeriesFold::new(); + assert_eq!(f.last, 0.0); + assert_eq!(f.max_drawdown, 0.0); + assert_eq!(f.sign_flips, 0); + } + + #[test] + fn series_fold_tracks_last_and_drawdown() { + let mut f = SeriesFold::new(); + for v in [10.0, 12.0, 8.0, 9.0] { + f.fold(v); + } + assert_eq!(f.last, 9.0); + assert_eq!(f.max_drawdown, 4.0); // peak 12 - trough 8 + assert_eq!(f.sign_flips, 0); // all positive + } + + #[test] + fn series_fold_counts_sign_flips_zero_distinct() { + let mut f = SeriesFold::new(); + for v in [1.0, -1.0, 0.0, 2.0] { + f.fold(v); + } + // signs + - 0 + -> flips at -, 0, + = 3 + assert_eq!(f.sign_flips, 3); + } +} +``` + +- [ ] **Step 2: Wire the module + export in lib.rs** + +In `crates/aura-core/src/lib.rs`, add `mod series_fold;` to the `mod` block +(after `mod scalar;` at line 37) and `pub use series_fold::SeriesFold;` to the +re-export block (after line 47): + +```rust +mod any; +mod cell; +mod column; +mod ctx; +mod error; +mod node; +mod scalar; +mod series_fold; + +pub use any::AnyColumn; +pub use cell::Cell; +pub use column::{Column, Window}; +pub use ctx::Ctx; +pub use error::KindMismatch; +pub use node::{ + zip_params, FieldSpec, Firing, Node, NodeSchema, ParamSpec, PortSpec, PrimitiveBuilder, +}; +pub use scalar::{Scalar, ScalarKind, Timestamp}; +pub use series_fold::SeriesFold; +``` + +- [ ] **Step 3: Run the tests** + +Run: `cargo test -p aura-core series_fold` +Expected: PASS — `series_fold_empty_is_zero`, `series_fold_tracks_last_and_drawdown`, `series_fold_counts_sign_flips_zero_distinct` all green. + +--- + +### Task 2: `Node::finalize` end-of-stream hook (aura-core) + +**Files:** +- Modify: `crates/aura-core/src/node.rs:280-298` + +- [ ] **Step 1: Write the failing test** + +In `crates/aura-core/src/node.rs`, inside `mod tests` (the `Bare` struct at +`node.rs:290` impls `Node` without `label`, relying on the default), add: + +```rust + #[test] + fn default_finalize_is_a_callable_noop() { + // a node that does not override finalize uses the default no-op; calling + // it compiles and does nothing (additive, non-breaking for existing nodes). + let mut b = Bare; + b.finalize(); + } +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: `cargo test -p aura-core default_finalize_is_a_callable_noop` +Expected: FAIL — compile error "no method named `finalize` found for struct `Bare`". + +- [ ] **Step 3: Add the default method to the trait** + +In `crates/aura-core/src/node.rs`, add after the `label` default method (closes +at line 282), before the trait's closing `}` at line 283: + +```rust + /// End-of-stream hook: `Harness::run` calls it once per node, in topological + /// order, after the source loop drains. A folding sink overrides it to flush + /// its accumulated summary; the default is a no-op, so existing nodes are + /// unaffected. Takes `&mut self` like `eval`, so `Node` stays object-safe. + fn finalize(&mut self) {} +``` + +- [ ] **Step 4: Run test to verify it passes** + +Run: `cargo test -p aura-core default_finalize_is_a_callable_noop` +Expected: PASS. + +--- + +### Task 3: `Harness::run` finalize epilogue (aura-engine) + +**Files:** +- Modify: `crates/aura-engine/src/harness.rs:443` (epilogue) and `mod tests` (probe + test) + +- [ ] **Step 1: Write the failing test** + +In `crates/aura-engine/src/harness.rs`, inside `mod tests`, add a probe node and +a test (mirrors the `boot(nodes, sigs, sources, edges)` + `h.run(..)` pattern of +`recorder_records_mixed_scalar_kinds`): + +```rust + struct FinalizeProbe { + id: usize, + tx: mpsc::Sender, + } + impl Node for FinalizeProbe { + fn lookbacks(&self) -> Vec { + vec![1] + } + fn eval(&mut self, _ctx: Ctx<'_>) -> Option<&[Cell]> { + None + } + fn finalize(&mut self) { + let _ = self.tx.send(self.id); + } + } + + #[test] + fn run_finalizes_every_node_once_after_the_stream() { + let (tx, rx) = mpsc::channel(); + // two sink probes, both fed by the single source; no inter-node edges. + let mut h = boot( + vec![ + Box::new(FinalizeProbe { id: 0, tx: tx.clone() }), + Box::new(FinalizeProbe { id: 1, tx }), + ], + vec![ + recorder_sig(&[ScalarKind::F64], Firing::Any), + recorder_sig(&[ScalarKind::F64], Firing::Any), + ], + vec![SourceSpec { + kind: ScalarKind::F64, + targets: vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }], + }], + vec![], + ) + .expect("valid"); + // nothing finalized before the run. + assert!(rx.try_recv().is_err()); + h.run(vec![Box::new(VecSource::new(f64_stream(&[(1, 1.0), (2, 2.0)])))]); + // each node finalized exactly once, after the stream drained. + let mut fired: Vec = rx.try_iter().collect(); + fired.sort_unstable(); + assert_eq!(fired, vec![0, 1]); + } +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: `cargo test -p aura-engine run_finalizes_every_node_once_after_the_stream` +Expected: FAIL — the assertion `fired == [0, 1]` fails (`fired` is empty: `run` +does not yet call `finalize`). (If `finalize` is not yet on the trait from Task +2, this is a compile error instead — Task 2 precedes this task.) + +- [ ] **Step 3: Insert the epilogue** + +In `crates/aura-engine/src/harness.rs`, the source `loop { ... }` closes at line +443. Insert, after the loop and before `run`'s closing `}` (line 444): + +```rust + // end-of-stream: flush every node once, in topological order, so folding + // sinks (C8) emit their accumulated summary. This runs AFTER the + // deterministic event loop, so it adds no within-sim concurrency (C1). + for &nidx in topo.iter() { + nodes[nidx].node.finalize(); + } +``` + +(`nodes` and `topo` are already in scope from the destructure +`let Harness { nodes, topo, out_edges, sources } = self;` at line 382.) + +- [ ] **Step 4: Run test to verify it passes** + +Run: `cargo test -p aura-engine run_finalizes_every_node_once_after_the_stream` +Expected: PASS. + +--- + +### Task 4: `summarize` drives `SeriesFold`; remove dead `sign0` (aura-engine) + +**Files:** +- Modify: `crates/aura-engine/src/report.rs:369-416` + +- [ ] **Step 1: Refactor `summarize` to drive `SeriesFold`** + +In `crates/aura-engine/src/report.rs`, add `SeriesFold` to the `aura_core` +import, then replace the body of `summarize` (lines 369-403) with: + +```rust +pub fn summarize( + equity: &[(Timestamp, f64)], + exposure: &[(Timestamp, f64)], +) -> RunMetrics { + // total_pips + max_drawdown fold over equity; sign_flips folds over exposure. + // SeriesFold is the single arithmetic source of truth (shared with the + // SeriesReducer sink), so this is byte-identical to the prior inline loops. + let mut eqf = SeriesFold::new(); + for &(_, v) in equity { + eqf.fold(v); + } + let mut exf = SeriesFold::new(); + for &(_, v) in exposure { + exf.fold(v); + } + RunMetrics { + total_pips: eqf.last, + max_drawdown: eqf.max_drawdown, + bias_sign_flips: exf.sign_flips, + r: None, + } +} +``` + +- [ ] **Step 2: Remove the now-dead `sign0`** + +Run: `rg -n "sign0" crates/aura-engine/src/report.rs` +Expected: only the definition (`fn sign0` at ~405-416) remains (no other caller). +Then delete the `fn sign0` definition and its doc comment (the block from `/// +Three-way sign:` through the closing `}` of `fn sign0`, report.rs ~405-416). + +- [ ] **Step 3: Verify summarize goldens stay byte-identical** + +Run: `cargo test -p aura-engine summarize` +Expected: PASS — all `summarize_*` tests green, including +`summarize_total_pips_is_last_cumulative_value`, +`summarize_max_drawdown_is_worst_peak_to_trough`, +`summarize_sign_flips_counts_signum_changes`, +`summarize_is_zero_on_empty_streams`, and every `summarize_r_*` test (unchanged). + +--- + +### Task 5: `GatedRecorder` filtering sink (aura-std) + +**Files:** +- Create: `crates/aura-std/src/gated_recorder.rs` +- Modify: `crates/aura-std/src/lib.rs:18-62` + +- [ ] **Step 1: Write the new sink with impl + failing tests** + +Create `crates/aura-std/src/gated_recorder.rs`: + +```rust +//! `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, + gate_col: usize, + last: Option<(Timestamp, Vec)>, + last_emitted: bool, + tx: Sender<(Timestamp, Vec)>, +} + +impl GatedRecorder { + pub fn new(kinds: &[ScalarKind], gate_col: usize, tx: Sender<(Timestamp, Vec)>) -> 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, + gate_col: usize, + firing: Firing, + tx: Sender<(Timestamp, Vec)>, + ) -> 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 { + 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)> { + 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()); + } +} +``` + +- [ ] **Step 2: Wire the module + export** + +In `crates/aura-std/src/lib.rs`, add `mod gated_recorder;` to the `mod` block +(18-38) and `pub use gated_recorder::GatedRecorder;` to the re-export block +(39-62, alongside `pub use recorder::Recorder;` at line 54). + +- [ ] **Step 3: Run the tests** + +Run: `cargo test -p aura-std gated_recorder` +Expected: PASS — `emits_gated_rows_and_flushes_a_nongated_final_row`, +`does_not_double_emit_a_gated_final_row`, `cold_warmup_records_nothing`. + +--- + +### Task 6: `SeriesReducer` folding sink (aura-std) + +**Files:** +- Create: `crates/aura-std/src/series_reducer.rs` +- Modify: `crates/aura-std/src/lib.rs:18-62` + +- [ ] **Step 1: Write the new sink with impl + failing tests** + +Create `crates/aura-std/src/series_reducer.rs`: + +```rust +//! `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)>, +} + +impl SeriesReducer { + pub fn new(tx: Sender<(Timestamp, Vec)>) -> Self { + Self { fold: SeriesFold::new(), last_ts: Timestamp(0), tx } + } + + pub fn builder(firing: Firing, tx: Sender<(Timestamp, Vec)>) -> 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 { + 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)> = 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)> = rx.try_iter().collect(); + assert_eq!(out, vec![(Timestamp(0), vec![Scalar::f64(0.0), Scalar::f64(0.0), Scalar::i64(0)])]); + } +} +``` + +- [ ] **Step 2: Wire the module + export** + +In `crates/aura-std/src/lib.rs`, add `mod series_reducer;` to the `mod` block and +`pub use series_reducer::SeriesReducer;` to the re-export block. + +- [ ] **Step 3: Run the tests** + +Run: `cargo test -p aura-std series_reducer` +Expected: PASS — `folds_and_emits_one_summary_row_on_finalize`, +`empty_finalize_emits_zero_summary`. + +--- + +### Task 7: Integration — folded == raw, and O(trades) memory (aura-engine tests) + +**Files:** +- Create: `crates/aura-engine/tests/streaming_reduction_equivalence.rs` +- Create: `crates/aura-engine/tests/streaming_reduction_memory.rs` + +- [ ] **Step 1: Write the equivalence test** + +Create `crates/aura-engine/tests/streaming_reduction_equivalence.rs`: + +```rust +//! Folded sink reductions are byte-for-byte equal to the raw-recorder path. +//! Drives the folding sinks node-by-node (the `stage1_r_e2e.rs` direct-node +//! style) over a synthetic dense R-record + equity/exposure series, and asserts +//! `summarize_r` over `GatedRecorder`'s emitted rows equals `summarize_r` over +//! the full record, and a `SeriesReducer` summary equals `summarize`'s fields. + +use aura_core::{AnyColumn, Ctx, Node, Scalar, ScalarKind, Timestamp}; +use aura_engine::{summarize, summarize_r}; +use aura_std::{GatedRecorder, SeriesReducer, PM_RECORD_KINDS, PM_WIDTH}; +use std::sync::mpsc; + +const CLOSED: usize = 0; +const REALIZED_R: usize = 1; +const OPEN: usize = 11; +const UNREALIZED_R: usize = 12; +const ENTRY_PRICE: usize = 6; +const STOP_PRICE: usize = 7; + +/// One dense R-record row. `closed`/`open` set the gate + the open-at-end flag; +/// the priced fields make `summarize_r`'s R arithmetic well-defined. +fn pm_row(closed: bool, realized: f64, open: bool, entry: f64, stop: f64) -> Vec { + let mut v = vec![Scalar::f64(0.0); PM_WIDTH]; + v[CLOSED] = Scalar::bool(closed); + v[REALIZED_R] = Scalar::f64(realized); + v[OPEN] = Scalar::bool(open); + v[UNREALIZED_R] = Scalar::f64(if open { realized } else { 0.0 }); + v[ENTRY_PRICE] = Scalar::f64(entry); + v[STOP_PRICE] = Scalar::f64(stop); + v +} + +#[test] +fn gated_recorder_then_summarize_r_equals_raw_summarize_r() { + // a realistic mix: holds, closed wins/losses, a hold tail, then an open-at-end row. + let full: Vec<(Timestamp, Vec)> = [ + pm_row(false, 0.0, false, 0.0, 0.0), + pm_row(true, 1.5, false, 100.0, 95.0), + pm_row(false, 0.0, false, 0.0, 0.0), + pm_row(true, -1.0, false, 100.0, 95.0), + pm_row(false, 0.0, false, 0.0, 0.0), + pm_row(false, 0.7, true, 100.0, 95.0), // open at end + ] + .into_iter() + .enumerate() + .map(|(i, r)| (Timestamp(i as i64 + 1), r)) + .collect(); + + // RAW: summarize_r over the whole dense record. + let raw = summarize_r(&full, 0.0); + + // FOLDED: drive GatedRecorder (gate = CLOSED), collect its emitted rows, fold. + let (tx, rx) = mpsc::channel(); + let mut g = GatedRecorder::new(&PM_RECORD_KINDS, CLOSED, tx); + let mut cols: Vec = + PM_RECORD_KINDS.iter().map(|&k| AnyColumn::with_capacity(k, 1)).collect(); + for (ts, row) in &full { + for (i, s) in row.iter().enumerate() { + cols[i].push(*s).unwrap(); + } + assert_eq!(g.eval(Ctx::new(&cols, *ts)), None); + } + g.finalize(); + let folded_rows: Vec<(Timestamp, Vec)> = rx.try_iter().collect(); + let folded = summarize_r(&folded_rows, 0.0); + + assert_eq!(folded, raw, "GatedRecorder+summarize_r must equal raw summarize_r"); +} + +#[test] +fn series_reducer_equals_summarize_pip_fields() { + let equity: Vec<(Timestamp, f64)> = + [10.0, 12.0, 8.0, 9.0, 7.0].iter().enumerate().map(|(i, &v)| (Timestamp(i as i64), v)).collect(); + let exposure: Vec<(Timestamp, f64)> = + [1.0, 1.0, -1.0, 0.0, 1.0].iter().enumerate().map(|(i, &v)| (Timestamp(i as i64), v)).collect(); + + let raw = summarize(&equity, &exposure); + + // FOLDED: one SeriesReducer per series; read pip fields from their summaries. + let fold_series = |series: &[(Timestamp, f64)]| -> Vec { + let (tx, rx) = mpsc::channel(); + let mut r = SeriesReducer::new(tx); + let mut col = vec![AnyColumn::with_capacity(ScalarKind::F64, 1)]; + for &(ts, v) in series { + col[0].push(Scalar::f64(v)).unwrap(); + r.eval(Ctx::new(&col, ts)); + } + r.finalize(); + rx.try_iter().next().expect("one summary row").1 + }; + let eqs = fold_series(&equity); + let exs = fold_series(&exposure); + + assert_eq!(eqs[0].as_f64(), raw.total_pips, "total_pips"); + assert_eq!(eqs[1].as_f64(), raw.max_drawdown, "max_drawdown"); + assert_eq!(exs[2].as_i64() as u64, raw.bias_sign_flips, "bias_sign_flips"); +} +``` + +- [ ] **Step 2: Run the equivalence test** + +Run: `cargo test -p aura-engine --test streaming_reduction_equivalence` +Expected: PASS — both `gated_recorder_then_summarize_r_equals_raw_summarize_r` +and `series_reducer_equals_summarize_pip_fields`. + +- [ ] **Step 3: Write the memory capstone test (own binary, counting allocator)** + +Create `crates/aura-engine/tests/streaming_reduction_memory.rs`: + +```rust +//! BLOCKER #138 capstone: the R-reduction's peak retained memory is +//! O(trades)+O(1), independent of cycle count. Drives `GatedRecorder` (gate = +//! CLOSED) through its real `mpsc` channel with a FIXED small number of closed +//! trades but a VARIABLE large number of hold cycles, generated LAZILY (one row +//! at a time, never all at once), and measures peak live bytes with a counting +//! global allocator. A retain-everything sink's peak would scale with the cycle +//! count; the folding sink's does not. Sole test in this binary, so the global +//! peak counter is not polluted by other tests running in parallel. + +use std::alloc::{GlobalAlloc, Layout, System}; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::mpsc; + +use aura_core::{AnyColumn, Ctx, Node, Scalar, ScalarKind, Timestamp}; +use aura_std::{GatedRecorder, PM_RECORD_KINDS, PM_WIDTH}; + +struct CountingAlloc; +static LIVE: AtomicUsize = AtomicUsize::new(0); +static PEAK: AtomicUsize = AtomicUsize::new(0); + +unsafe impl GlobalAlloc for CountingAlloc { + unsafe fn alloc(&self, layout: Layout) -> *mut u8 { + let p = unsafe { System.alloc(layout) }; + if !p.is_null() { + let now = LIVE.fetch_add(layout.size(), Ordering::Relaxed) + layout.size(); + PEAK.fetch_max(now, Ordering::Relaxed); + } + p + } + unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) { + LIVE.fetch_sub(layout.size(), Ordering::Relaxed); + unsafe { System.dealloc(ptr, layout) } + } +} + +#[global_allocator] +static A: CountingAlloc = CountingAlloc; + +fn reset_peak() { + PEAK.store(LIVE.load(Ordering::Relaxed), Ordering::Relaxed); +} +fn peak() -> usize { + PEAK.load(Ordering::Relaxed) +} + +const CLOSED: usize = 0; +const REALIZED_R: usize = 1; +const K_TRADES: usize = 8; + +/// Peak live bytes to drive `GatedRecorder` over `K_TRADES` closed trades, each +/// followed by `holds` hold cycles, sourced lazily (one row built at a time), +/// then fold the emitted rows. With a folding sink the peak is O(trades). +fn peak_bytes(holds: usize) -> usize { + reset_peak(); + let (tx, rx) = mpsc::channel(); + let mut g = GatedRecorder::new(&PM_RECORD_KINDS, CLOSED, tx); + let mut cols: Vec = + PM_RECORD_KINDS.iter().map(|&k| AnyColumn::with_capacity(k, 1)).collect(); + let mut ts = 0i64; + let mut push = |g: &mut GatedRecorder, cols: &mut Vec, ts: &mut i64, closed: bool, r: f64| { + let mut row = vec![Scalar::f64(0.0); PM_WIDTH]; + row[CLOSED] = Scalar::bool(closed); + row[REALIZED_R] = Scalar::f64(r); + for (i, s) in row.iter().enumerate() { + cols[i].push(*s).unwrap(); + } + g.eval(Ctx::new(cols, Timestamp(*ts))); + *ts += 1; + }; + for k in 0..K_TRADES { + let r = if k % 2 == 0 { 1.0 } else { -0.5 }; + push(&mut g, &mut cols, &mut ts, true, r); + for _ in 0..holds { + push(&mut g, &mut cols, &mut ts, false, 0.0); + } + } + g.finalize(); + let rows: Vec<(Timestamp, Vec)> = rx.try_iter().collect(); + assert_eq!(rows.len(), K_TRADES, "only the closed trades are retained"); + let p = peak(); + drop(rows); + p +} + +#[test] +fn r_reduction_peak_memory_is_independent_of_cycle_count() { + const SMALL_HOLDS: usize = 64; // 8 * (1+64) = 520 cycles + const LARGE_HOLDS: usize = 50_000; // 8 * (1+50000) = 400_008 cycles + + let small = peak_bytes(SMALL_HOLDS); + let large = peak_bytes(LARGE_HOLDS); + + // cycle count grew ~770x; an O(trades)+O(1) sink holds its peak essentially + // flat. Generous 4x headroom so only genuine O(cycles) retention trips it. + assert!( + large <= small.saturating_mul(4), + "R-reduction peak scales with cycle count (O(cycles) retention): \ + small {} cycles -> {} bytes; large {} cycles -> {} bytes ({:.0}x). \ + The folding sink must retain O(trades)+O(1).", + SMALL_HOLDS * (1 + 0) + K_TRADES * (1 + SMALL_HOLDS), + small, + K_TRADES * (1 + LARGE_HOLDS), + large, + large as f64 / small.max(1) as f64, + ); +} +``` + +- [ ] **Step 4: Run the memory capstone** + +Run: `cargo test -p aura-engine --test streaming_reduction_memory` +Expected: PASS — `r_reduction_peak_memory_is_independent_of_cycle_count` (peak +for 400k cycles is within 4x of peak for 520 cycles, same 8 trades). + +--- + +### Task 8: Wire the reducers into the sweep; thread `reduce` (aura-cli) + +**Files:** +- Modify: `crates/aura-cli/src/main.rs:1924-1988` (`stage1_r_graph` gains `reduce`) +- Modify: `crates/aura-cli/src/main.rs:1179, 1206-1242, 2006` (thread `reduce`; folded consumer) + +- [ ] **Step 1: Add `reduce: bool` to `stage1_r_graph` and branch the sinks** + +In `crates/aura-cli/src/main.rs`, add the import (alongside the existing aura-std +import at line 31): `GatedRecorder`, `SeriesReducer`. Change the signature +(line 1931) to add a trailing `reduce: bool` param. Replace the sink-wiring +block (the eq/ex Recorders at 1949-1950, the rrec at 1958, and the +`r_equity`/`req` block at 1959-1986) so that: + +- `eq`/`ex`: when `reduce`, `SeriesReducer::builder(Firing::Any, tx_eq)` / + `(.., tx_ex)`; else the current `Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx_eq)` / `(.., tx_ex)`. +- `rrec`: when `reduce`, `GatedRecorder::builder(PM_RECORD_KINDS.to_vec(), gate_col, Firing::Any, tx_r)` where + `let gate_col = PM_FIELD_NAMES.iter().position(|&n| n == "closed_this_cycle").expect("PM record has a closed_this_cycle column");` + else the current `Recorder::builder(PM_RECORD_KINDS.to_vec(), Firing::Any, tx_r)`. +- the dense R-record tap loop (1980-1982) connects `exec.output(field)` to + `rrec.input(COL_PORTS[i])` in BOTH modes (GatedRecorder declares the same + per-kind input ports as Recorder). +- the `r_equity` LinComb node + `req` Recorder + their connects (1959-1965, + 1984-1986) are wired ONLY when `!reduce` (the `r_equity` tap is unused without + `--trace`; `tx_req` is simply dropped when `reduce`). + +The shared signal/broker/exec wiring (1933-1957, 1966-1979, 1983) stays as +today. Concretely the sink region becomes: + +```rust + let gate_col = PM_FIELD_NAMES + .iter() + .position(|&n| n == "closed_this_cycle") + .expect("PM record has a closed_this_cycle column"); + let eq = if reduce { + g.add(SeriesReducer::builder(Firing::Any, tx_eq)) + } else { + g.add(Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx_eq)) + }; + let ex = if reduce { + g.add(SeriesReducer::builder(Firing::Any, tx_ex)) + } else { + g.add(Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx_ex)) + }; + let exec = g.add(if stop_open { + risk_executor_vol_open(1.0) + } else { + risk_executor(StopRule::Vol { length: STAGE1_R_STOP_LENGTH, k: STAGE1_R_STOP_K }, 1.0) + }); + let rrec = if reduce { + g.add(GatedRecorder::builder(PM_RECORD_KINDS.to_vec(), gate_col, Firing::Any, tx_r)) + } else { + g.add(Recorder::builder(PM_RECORD_KINDS.to_vec(), Firing::Any, tx_r)) + }; + let price = g.source_role("price", ScalarKind::F64); + g.feed( + price, + [fast.input("series"), slow.input("series"), broker.input("price"), exec.input("price")], + ); + g.connect(fast.output("value"), spread.input("lhs")); + g.connect(slow.output("value"), spread.input("rhs")); + g.connect(spread.output("value"), exposure.input("signal")); + g.connect(exposure.output("bias"), broker.input("exposure")); + g.connect(exposure.output("bias"), ex.input("col[0]")); + g.connect(exposure.output("bias"), exec.input("bias")); + g.connect(broker.output("equity"), eq.input("col[0]")); + for (i, field) in PM_FIELD_NAMES.iter().enumerate() { + g.connect(exec.output(field), rrec.input(COL_PORTS[i].as_str())); + } + if !reduce { + let r_equity = g.add( + LinComb::builder(2) + .bind("weights[0]", Scalar::f64(1.0)) + .bind("weights[1]", Scalar::f64(1.0)), + ); + let req = g.add(Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx_req)); + g.connect(exec.output("cum_realized_r"), r_equity.input("term[0]")); + g.connect(exec.output("unrealized_r"), r_equity.input("term[1]")); + g.connect(r_equity.output("value"), req.input("col[0]")); + } + g.build().expect("stage1_r wiring resolves") +``` + +(The `broker`/`fast`/`slow`/`spread`/`exposure` `g.add(..)` calls above this +block are unchanged; only the eq/ex/rrec adds move into the `if reduce` form and +the r_equity/req block becomes conditional. The `move`d `tx_req` is unused when +`reduce` — that is intended; do not add a sink for it.) + +- [ ] **Step 2: Thread `reduce` through the three call sites** + +- `main.rs:1179` (throwaway param-space build): append `, true` → `stage1_r_graph(tx_eq, tx_ex, tx_r, tx_req, None, None, true, true)` (reduce does not affect `param_space`; the varying axes live on the SMA/vol-stop nodes present in both modes). +- `main.rs:1206-1242` (the `.sweep(|point| { .. })` closure): replace the body so it computes `reduce` once and branches the consumer. Current lines 1208-1239 become: + +```rust + .sweep(|point| { + let (tx_eq, rx_eq) = mpsc::channel(); + let (tx_ex, rx_ex) = mpsc::channel(); + let (tx_r, rx_r) = mpsc::channel(); + let (tx_req, rx_req) = mpsc::channel(); + let reduce = trace.is_none(); + let mut h = stage1_r_graph(tx_eq, tx_ex, tx_r, tx_req, None, None, true, reduce) + .bootstrap_with_cells(point) + .expect("stage1-r grid points are kind-checked against param_space"); + h.run(data.run_sources()); + let mut named: Vec<(String, Scalar)> = zip_params(&space, point) + .into_iter() + .map(|(n, v)| (stage1_r_friendly_name(&n), v)) + .collect(); + named.push(("bias_scale".to_string(), Scalar::f64(0.5))); + let key = member_key(&named, &varying); + let mut manifest = sim_optimal_manifest(named, window, 0, pip); + manifest.broker = stage1_r_broker_label(pip); + let metrics = if reduce { + // folded: GatedRecorder emits O(trades) R rows; each SeriesReducer + // emits one [last, max_drawdown, sign_flips] summary row. + let r_rows: Vec<(Timestamp, Vec)> = rx_r.try_iter().collect(); + let (total_pips, max_drawdown) = rx_eq + .try_iter() + .next() + .map(|(_, row)| (row[0].as_f64(), row[1].as_f64())) + .unwrap_or((0.0, 0.0)); + let bias_sign_flips = + rx_ex.try_iter().next().map(|(_, row)| row[2].as_i64() as u64).unwrap_or(0); + let mut m = RunMetrics { total_pips, max_drawdown, bias_sign_flips, r: None }; + m.r = Some(summarize_r(&r_rows, 0.0)); + m + } else { + // trace path (--trace set): raw recorders, persist, summarize — today's code. + let eq_rows: Vec<(Timestamp, Vec)> = rx_eq.try_iter().collect(); + let ex_rows: Vec<(Timestamp, Vec)> = rx_ex.try_iter().collect(); + let r_rows: Vec<(Timestamp, Vec)> = rx_r.try_iter().collect(); + let req_rows: Vec<(Timestamp, Vec)> = rx_req.try_iter().collect(); + if let Some(name) = trace { + persist_traces_r(&format!("{name}/{key}"), &manifest, &eq_rows, &ex_rows, &req_rows); + } + let mut m = summarize(&f64_field(&eq_rows, 0), &f64_field(&ex_rows, 0)); + m.r = Some(summarize_r(&r_rows, 0.0)); + m + }; + RunReport { manifest, metrics } + }) +``` + +- `main.rs:2006` (`run_stage1_r`, the single-run reference path): append `, false` + → `stage1_r_graph(tx_eq, tx_ex, tx_r, tx_req, Some(2), Some(4), false, false)`. + The rest of `run_stage1_r` (collect eq/ex/r/req + persist + summarize) is + unchanged. + +- [ ] **Step 3: Verify the existing stage1-r goldens stay byte-identical** + +Run: `cargo test -p aura-cli` +Expected: PASS — every existing aura-cli test, including the stage1-r single-run, +the stage1-r sweep grid tests (`sweep_strategy_stage1_r_*`), and any `--trace` +test, with byte-identical reported metrics (the folded no-trace path produces the +same values the raw path did, by the Task 7 equivalence proof). + +- [ ] **Step 4: Full workspace + lint gate** + +Run: `cargo test --workspace` +Expected: PASS — all crates green (the new tests plus every pre-existing test). + +Run: `cargo clippy --workspace --all-targets -- -D warnings` +Expected: PASS — no warnings (in particular, no dead-code warning for a leftover +`sign0`, and no unused-import warning).