Aura/docs/specs/0070-streaming-sink-reductions.md

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<Scalar>)) — 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 SeriesReducers (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 Recorders 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)

$ 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)

// 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

// 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<f64>,    // 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);
    }
}

// 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)

// crates/aura-std/src/gated_recorder.rs — Recorder that filters to O(trades) rows
struct GatedRecorder {
    kinds: Vec<ScalarKind>,
    gate_col: usize,                  // emit a row only when row[gate_col] is true
    last: Option<(Timestamp, Vec<Scalar>)>,   // the genuine final row
    last_emitted: bool,               // was `last` already emitted via the gate?
    tx: Sender<(Timestamp, Vec<Scalar>)>,
}
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); }
        }
    }
}

// crates/aura-std/src/series_reducer.rs — folds one f64 series, flushes once
struct SeriesReducer { fold: SeriesFold, warm: bool, tx: Sender<(Timestamp, Vec<Scalar>)> }
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)

// crates/aura-engine/src/harness.rs ~371 — after `loop { ... }` breaks:
for &nidx in topo.iter() { nodes[nidx].node.finalize(); }

// 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 Recorders → .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.