From ffd18b98ea3d35a5fb930302e37c983166d9f256 Mon Sep 17 00:00:00 2001 From: Brummel Date: Thu, 25 Jun 2026 11:29:51 +0200 Subject: [PATCH] =?UTF-8?q?audit(0070):=20cycle=20close=20=E2=80=94=20lift?= =?UTF-8?q?=20finalize=20lifecycle=20to=20ledger=20C8,=20drop=20ephemera?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Cycle 0070 (streaming sink reductions, #138) close. Durable rationale lifted to the ledger: C8 now records the end-of-stream `finalize` lifecycle (the second node phase beside `eval`), its `Harness::run` topo-order epilogue, and its C1/C7/C8 reconciliation — with the open `Recorder`->`Probe` / per-cycle-allocation question (#77) cross-referenced, since `finalize` is the "non-channel exit" that issue's accumulate-then-read option named as missing. The per-cycle spec and plan are git-rm'd (ephemeral, valid only for their cycle). Audit (architect) found no invariant breakage; noted debt, left as-is: the two-mode `stage1_r_graph` (reduce vs raw sink wiring) rests on the CLI-seam equivalence test rather than structural unification, and `SeriesReducer.last_ts` carries the last *warm* cycle's ts (latent only if `--trace` ever folds). refs #138 --- docs/design/INDEX.md | 20 + docs/plans/0070-streaming-sink-reductions.md | 1065 ------------------ docs/specs/0070-streaming-sink-reductions.md | 301 ----- 3 files changed, 20 insertions(+), 1366 deletions(-) delete mode 100644 docs/plans/0070-streaming-sink-reductions.md delete mode 100644 docs/specs/0070-streaming-sink-reductions.md diff --git a/docs/design/INDEX.md b/docs/design/INDEX.md index 643cf1e..0842da3 100644 --- a/docs/design/INDEX.md +++ b/docs/design/INDEX.md @@ -275,6 +275,26 @@ out-of-graph side effect. In-graph routing stays engine-owned data (the edge tab of the graph is the node's own side effect — and that boundary is the determinism / graph-as-data boundary (C1/C7). +**Refinement (cycle 0070 — end-of-stream `finalize` lifecycle, 2026-06-25).** The +node contract gains a second lifecycle phase beside `eval`: `finalize(&mut self)`, +a **default-no-op** trait method the engine calls **once per node, in topological +order, after the source loop drains** (the `Harness::run` epilogue). It lets a sink +**fold online** — accumulate per `eval` into O(trades)/O(1) owned state and flush +one compact summary at stream end — instead of pushing a record every fired cycle +into an unbounded channel that buffers O(cycles) rows until the run ends (the +retention BLOCKER #138 hit: a full-history Stage-1 sweep held ~2 GiB/member over +~5.5M one-minute bars). C1/C7/C8 are preserved: `finalize` runs once *after* the +deterministic event loop, adding no within-sim concurrency (C1); a folding sink +holds only owned accumulator state, no interior mutability (C7); it still declares +`output: vec![]` and its flush is the same out-of-graph side effect (C8) — one +summary row, not a per-cycle stream. Realized by `GatedRecorder` (emits only gated +rows + the genuine final row) and `SeriesReducer` (folds one column, emits one +summary row), aura-std siblings of the per-cycle `Recorder`, which survives for the +live / `--trace` / test-tap path. The recording sink's *own* per-cycle allocation +(the `Recorder`→`Probe` rename + its accumulate-vs-stream choice, #77) stays open; +`finalize` is the "non-channel exit from the graph" that issue's +accumulate-then-read option named as missing. + **Refinement (Construction-layer milestone — render labels, 2026-06-05).** A node additionally exposes `label() -> String`, a **single-line, non-load-bearing** render symbol: a default trait method the run loop never calls (wiring is by diff --git a/docs/plans/0070-streaming-sink-reductions.md b/docs/plans/0070-streaming-sink-reductions.md deleted file mode 100644 index 2854caa..0000000 --- a/docs/plans/0070-streaming-sink-reductions.md +++ /dev/null @@ -1,1065 +0,0 @@ -# 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). diff --git a/docs/specs/0070-streaming-sink-reductions.md b/docs/specs/0070-streaming-sink-reductions.md deleted file mode 100644 index 1f6d64a..0000000 --- a/docs/specs/0070-streaming-sink-reductions.md +++ /dev/null @@ -1,301 +0,0 @@ -# Streaming Sink Reductions (BLOCKER #138) — Design Spec - -**Date:** 2026-06-25 -**Status:** Draft — awaiting user spec review -**Authors:** orchestrator + Claude - -## Goal - -A full-history `stage1-r` parameter sweep must run in **bounded memory per -member** — O(trades)+O(1), independent of the cycle count — so that -multi-member full-history sweeps (and, downstream, walk-forward, #139) no -longer exhaust RAM+swap and SIGKILL-137. - -Today each swept member holds **~2 GiB** over ~5.5M one-minute cycles. The -dominant retainer is the unbounded `mpsc` channel each `Recorder` sink fills -**during** `Harness::run` (one row per fired cycle, nothing draining), plus a -second O(cycles) copy when `run_one` `.collect()`s it. The dense R-record -(14 columns × ~5.5M cycles ≈ 1.4 GB) dominates; the equity/exposure taps add -~0.5 GB; the `r_equity` tap is collected then **discarded** in the no-trace -sweep — pure waste. - -This is **M3** (the mechanism the user chose for #138): an end-of-stream -`finalize` lifecycle on sinks lets a sink **fold online into O(trades)+O(1) -state and flush a compact summary once**, instead of streaming every per-cycle -row out to a channel that buffers them all. - -Out of scope: the realistic-broker / cost layer (Stage 2); the -`--trace` persistence path (it writes full per-cycle streams to disk — -inherently O(cycles), kept as today, an explicit opt-in overlay, not the -BLOCKER); the SMA sample and `walkforward`/`mc` CLI paths (untouched here; -#139 builds on this). - -## Architecture - -Three layers, smallest-blast-radius first: - -1. **`aura-core` — the `finalize` lifecycle hook + a shared online fold.** - - `Node::finalize(&mut self) {}` — an **additive default-no-op** method on - the `Node` trait. Every existing node keeps compiling untouched; only a - folding sink overrides it. Object-safe (`&mut self`, no generics), like - `eval`. - - `SeriesFold` — a single-series online accumulator producing exactly the - three pip statistics `summarize` computes (`last`, `max_drawdown`, - `sign_flips`). One arithmetic implementation, in the crate **both** - `aura-engine` (`summarize`) and `aura-std` (the pip reducer sink) already - depend on — so there is **no dependency inversion** and the folded path is - **byte-identical to `summarize` by construction**, not merely by a pinned - test. - -2. **`aura-engine` — call `finalize` at end-of-stream; drive the shared fold.** - - `Harness::run` gains a **post-loop epilogue**: after the source loop - breaks (all sources exhausted), call `finalize()` on every node once, in - topological order. This runs **after** the deterministic event loop, so it - adds **no within-sim concurrency** (C1 intact). - - `summarize` is refactored to drive `SeriesFold` (one instance over the - equity series for `last`+`max_drawdown`, one over the exposure series for - `sign_flips`), keeping its public signature and its result byte-identical. - `summarize_r` is **unchanged**. - -3. **`aura-std` — two folding sink nodes (siblings of `Recorder`).** Both emit - the **same channel type** as `Recorder` (`(Timestamp, Vec)`) — no - metric types cross into `aura-std` — just compacted: - - `GatedRecorder` — a `Recorder` that emits a row only when a designated - **gate column** (a `bool`) is true, and **always** emits the genuine final - row once on `finalize` (deduped against the last gated emit). Fed the dense - R-record gated on the `closed` column, it yields the **O(trades)** closed - rows + the final row — exactly the rows `summarize_r` extracts — so - `summarize_r` consumes them **verbatim** (zero golden risk). - - `SeriesReducer` — a sink over one `f64` column that folds a `SeriesFold` - per `eval` and, on `finalize`, emits **one** row - `[last, max_drawdown, sign_flips]`. Used twice (equity, exposure); the - consumer reads `last`+`max_drawdown` from the equity instance and - `sign_flips` from the exposure instance. - -4. **`aura-cli` — wire the reducers; drain O(trades) instead of O(cycles).** - - `stage1_r_graph` gains a `reduce: bool`. When `reduce` (the no-trace sweep, - the BLOCKER path): wire `GatedRecorder` (R) + two `SeriesReducer`s - (equity, exposure); **omit** the `r_equity` tap entirely (it is unused - without `--trace`). When `!reduce` (today's `--trace` path and the single - `run_stage1_r`): wire the raw `Recorder`s exactly as today. - - The no-trace `run_one` builds `RunReport` from the folded rows: - `summarize_r(&gated_r_rows)` for the R block; the two single-summary rows - for the pip block. The `--trace` branch and `run_stage1_r` are **unchanged** - (they keep the raw recorders + `summarize`/`summarize_r` over full rows — - the byte-for-byte reference the equivalence test compares against). - -The single feedback-free invariant set is preserved: C1 (finalize is a -post-loop, single-threaded pass), C7 (the sinks hold only owned accumulator -state + an owned `Sender` — no `Rc`/`RefCell`, like `Recorder`), C8 (the sinks -declare empty output and emit nothing downstream). - -## Concrete code shapes - -### User-facing evidence (infra cycle — the invocation that must stop OOM-ing) - -```console -$ aura sweep --strategy stage1-r --real EURUSD \ - --fast 240,480,1920 --slow 1920,7680 --stop-length 1920 --stop-k 2.0 -# 6-member grid over ~5.5M one-minute cycles, no --trace. -# BEFORE: peak RSS ≈ members × ~2 GiB → swap exhaustion → SIGKILL (137). -# AFTER: peak RSS ≈ members × O(trades) → completes; metrics byte-identical. -``` - -The cycle delivers no new user surface; its empirical evidence is (a) the -must-fail memory fixture below and (b) the equivalence fixture proving the -metrics are unchanged. - -### The lifecycle hook (aura-core) - -```rust -// crates/aura-core/src/node.rs — additive, default no-op -trait Node { - fn eval(&mut self, ctx: Ctx) -> Option<&[Cell]>; // unchanged - // NEW: end-of-stream flush. default no-op → existing nodes untouched. - fn finalize(&mut self) {} -} -``` - -### The shared online fold (aura-core) — single source of truth - -```rust -// crates/aura-core/src/... — the exact three folds `summarize` performs today -struct SeriesFold { - last: f64, // total_pips = last value seen (0.0 if none) - peak: f64, // running max, seeded NEG_INFINITY (as summarize) - max_drawdown: f64, // running max(peak - v), >= 0.0 - prev_sign: Option, // sign0 of the previous value - sign_flips: u64, -} -impl SeriesFold { - fn fold(&mut self, v: f64) { - // operations and order kept identical to summarize's loops, so the - // result is bit-for-bit equal (IEEE-754 non-associativity respected). - 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 { if s != p { self.sign_flips += 1; } } - self.prev_sign = Some(s); - } -} -``` - -```rust -// crates/aura-engine/src/report.rs — summarize now drives SeriesFold, signature unchanged -pub fn summarize(equity: &[(Timestamp, f64)], exposure: &[(Timestamp, f64)]) -> RunMetrics { - let mut eqf = SeriesFold::new(); - for &(_, v) in equity { eqf.fold(v); } // total_pips, max_drawdown - let mut exf = SeriesFold::new(); - for &(_, v) in exposure { exf.fold(v); } // sign_flips - RunMetrics { total_pips: eqf.last, max_drawdown: eqf.max_drawdown, - bias_sign_flips: exf.sign_flips, r: None } -} -// summarize_r: UNCHANGED. -``` - -### The folding sinks (aura-std) - -```rust -// crates/aura-std/src/gated_recorder.rs — Recorder that filters to O(trades) rows -struct GatedRecorder { - kinds: Vec, - gate_col: usize, // emit a row only when row[gate_col] is true - last: Option<(Timestamp, Vec)>, // the genuine final row - last_emitted: bool, // was `last` already emitted via the gate? - tx: Sender<(Timestamp, Vec)>, -} -impl Node for GatedRecorder { - fn eval(&mut self, ctx: Ctx) -> Option<&[Cell]> { - let row = read_declared_columns(ctx)?; // None during warm-up, as Recorder - 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 // sink: emits nothing downstream (C8) - } - fn finalize(&mut self) { - // ensure summarize_r's `record.last()` sees the true final row, without - // double-counting a final row that was already emitted as a closed trade. - if !self.last_emitted { - if let Some(row) = self.last.take() { let _ = self.tx.send(row); } - } - } -} -``` - -```rust -// crates/aura-std/src/series_reducer.rs — folds one f64 series, flushes once -struct SeriesReducer { fold: SeriesFold, warm: bool, tx: Sender<(Timestamp, Vec)> } -impl Node for SeriesReducer { - 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.warm = true; - None - } - fn finalize(&mut self) { - // one summary row; ts is the final cycle's — carried for join symmetry, unused by the consumer. - 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.final_ts(), row)); - } -} -``` - -### The run loop epilogue (aura-engine) and the folded consumer (aura-cli) - -```rust -// crates/aura-engine/src/harness.rs ~371 — after `loop { ... }` breaks: -for &nidx in topo.iter() { nodes[nidx].node.finalize(); } -``` - -```rust -// crates/aura-cli/src/main.rs — no-trace run_one (the BLOCKER path) -h.run(data.run_sources()); // channels now fill only at finalize -let r_rows: Vec<_> = rx_r.try_iter().collect(); // O(trades): cheap -let eq = rx_eq.try_iter().next().unwrap_or(empty_summary()); // one folded row -let ex = rx_ex.try_iter().next().unwrap_or(empty_summary()); -let mut metrics = RunMetrics { - total_pips: eq.1[0].as_f64(), max_drawdown: eq.1[1].as_f64(), - bias_sign_flips: ex.1[2].as_i64() as u64, r: None, -}; -metrics.r = Some(summarize_r(&r_rows, 0.0)); // summarize_r unchanged -``` - -## Components - -| Component | Crate | Change | -|-----------|-------|--------| -| `Node::finalize` | aura-core | NEW default-no-op trait method | -| `SeriesFold` | aura-core | NEW online accumulator (the three pip folds) | -| `Harness::run` | aura-engine | post-loop `finalize()` epilogue (topo order) | -| `summarize` | aura-engine | refactor to drive `SeriesFold`; signature + result unchanged | -| `summarize_r` | aura-engine | UNCHANGED | -| `GatedRecorder` | aura-std | NEW filtering sink (gate column + final-row flush) | -| `SeriesReducer` | aura-std | NEW single-series folding sink | -| `Recorder` | aura-std | UNCHANGED (kept for `--trace` + SMA + tests) | -| `stage1_r_graph` | aura-cli | `reduce: bool`; wire reducers / omit `r_equity` when reducing | -| `run_one` (sweep) | aura-cli | no-trace branch builds `RunReport` from folded rows | -| `run_stage1_r`, `--trace` branch | aura-cli | UNCHANGED (reference path) | - -## Data flow - -`reduce` (no-trace sweep): -`price → graph → {GatedRecorder(R, gate=closed), SeriesReducer(equity), -SeriesReducer(exposure)}`. During `run`: per-cycle folds, nothing buffered -beyond O(trades) closed rows. At `finalize`: GatedRecorder flushes the final -row; each SeriesReducer flushes one summary row. `run_one` drains O(trades)+2 -rows → `summarize_r` + direct `RunMetrics` build. - -`!reduce` (`--trace`, single run): unchanged — raw `Recorder`s → -`.collect()` → `persist_traces_r` (trace) + `summarize`/`summarize_r`. - -## Error handling - -No new failure modes. A reducer sink whose declared column is missing/wrong -kind panics at the same "checked at wiring" boundary `Recorder`/`f64_field` use -today. `try_iter().next()` on an empty folded channel (a run with zero warm -cycles) falls back to the zero-summary, matching `summarize` on empty input -(`total_pips=0.0`, `max_drawdown=0.0`, `flips=0`). - -## Testing strategy - -- **End-to-end RED (the BLOCKER pin).** A counting-`#[global_allocator]` - integration test driving a real `Harness` + the folding sinks over a - synthetic stream with a FIXED small number of trades but a VARIABLE large - cycle count (e.g. 520 vs ~400k cycles): assert peak retained bytes through - the **recorder → channel → reduction** path does **not** scale with cycle - count (O(trades)+O(1)). Fails on today's `Recorder`+`.collect()` path; passes - with the folding sinks. (Reference scaffolding preserved at - `scratchpad/r_reduction_streaming.red-reference.rs` — but it must exercise - the channel path end-to-end, not `summarize_r` in isolation.) -- **Golden equivalence (byte-for-byte).** On the same synthetic + a real - fixture window, assert the folded `RunReport` (R block + pip block) equals - the raw-recorder `summarize`/`summarize_r` `RunReport` **field-for-field**. - This is the proof the folded path changed nothing observable. -- **`finalize` lifecycle unit tests** (aura-engine): `run` calls `finalize` - once per node after the stream ends, in topo order; a default node's - `finalize` is a no-op. -- **`GatedRecorder` unit tests** (aura-std): emits only gated rows; emits the - final row on `finalize`; does **not** double-emit when the final row was - gated-true. -- **`SeriesReducer` unit tests** (aura-std): folds match `SeriesFold`; one row - on `finalize`; nothing per-eval. -- **Existing goldens** (R metrics, pip metrics, sweep family, single run, - `--trace`): all must stay green and byte-identical. - -## Acceptance criteria - -1. A no-trace `stage1-r` sweep's **peak retained memory per member is - O(trades)+O(1)**, independent of cycle count (the RED test passes). -2. Every existing metric value (R and pip, single run and sweep) is - **byte-identical** (the equivalence test + all existing goldens pass). -3. `Node::finalize` is additive and default-no-op; no existing node is edited - to satisfy the compiler. -4. C1 (no within-sim concurrency), C7 (owned accumulator state, no interior - mutability), C8 (sinks emit nothing) are preserved. -5. `cargo test --workspace` and `cargo clippy --workspace --all-targets -- -D - warnings` are green.