Aura/docs/plans/0071-stage1-breakout.md

Stage-1 Donchian Breakout Candidate — Implementation Plan

Parent spec: docs/specs/0071-stage1-breakout.md

For agentic workers: REQUIRED SUB-SKILL: use the implement skill to run this plan. Steps use - [ ] checkboxes for tracking.

Goal: Add a channel-breakout signal candidate (--strategy stage1-breakout) that screens under the same Stage-1 R yardstick as stage1-r, swapping only the signal leg (SMA-cross → Donchian channel direction).

Architecture: Two new aura-std nodes (RollingMax/RollingMin, sliding-window extremum via a monotonic deque, mirroring the Sma/Delay node shape). A new stage1_breakout_graph mirrors stage1_r_graph but builds the signal leg close → Delay(1) → {RollingMax,RollingMin}(N) → {Gt,Gt} → {Latch,Latch} → Sub = bias ∈ {−1,0,+1}, feeding the identical bias seam; the vol-stop defines R unchanged. A stage1_breakout_sweep_family iterates the channel × stop grid manually with fully-bound graphs (sidestepping parameter-ganging, since one channel drives two nodes). A Strategy::Stage1Breakout variant + --channel flag wire it to the CLI.

Tech Stack: aura-std (node primitives), aura-core (Node trait), aura-cli (strategy graph + sweep family + arg parse), aura-engine (GraphBuilder, Harness, SweepFamily/SweepPoint, summarize/summarize_r).

---

Files this plan creates or modifies:

• Create: crates/aura-std/src/rolling_max.rs — sliding-window max node (descending monotonic deque)
• Create: crates/aura-std/src/rolling_min.rs — sliding-window min node (ascending monotonic deque)
• Modify: crates/aura-std/src/lib.rs:18-66 — register the two modules + re-exports
• Create: crates/aura-engine/tests/stage1_breakout_e2e.rs — C2-causality + ±1-latch composition test
• Modify: crates/aura-cli/src/main.rs — imports (Gt/Latch/RollingMax/RollingMin, SweepPoint); stage1_breakout_graph; stage1_breakout_sweep_family; Strategy::Stage1Breakout (enum 1286, parse arm 1336, dispatch arm 1422); --channel flag (parse 1345) + Stage1RGrid.channel field (1123-1140)
• Test: crates/aura-cli/tests/cli_run.rs — aura sweep --strategy stage1-breakout seam: r block present + folded-no-trace == raw-trace

---

Task 1: RollingMax node

Files:

• Create: crates/aura-std/src/rolling_max.rs

• Modify: crates/aura-std/src/lib.rs:18-66

• Step 1: Write the node file with its failing tests

Create crates/aura-std/src/rolling_max.rs:

//! `RollingMax` — maximum over the last `length` values of one f64 input.
//!
//! Sliding-window maximum maintained by a **descending monotonic deque** (the
//! textbook O(1)-amortized sliding-window-max): each cycle pops from the back every
//! element <= the new sample (they can never again be the window max), pushes the new
//! one, and evicts the front once its stream index leaves the window. The front is
//! always the current window max. A per-cycle O(N) re-scan would be too slow at the
//! large `N` (channel lengths in the thousands of bars) the breakout screen sweeps.
//!
//! Like `Sma`/`Delay`, the window lives in node state, so `eval` reads only the newest
//! sample and `lookbacks()` is `1`. Warm-up is **skip-emit**: `None` until `length`
//! samples have passed (the window is not yet full). The deque is pre-sized to
//! `length` so the hot path is allocation-free (C7). Per the "operator is topology"
//! convention (see `gt.rs`), `RollingMax` and `RollingMin` are two node types, not one
//! node with a max/min param.

use aura_core::{
    Cell, Ctx, FieldSpec, Firing, Node, NodeSchema, ParamSpec, PortSpec, PrimitiveBuilder,
    ScalarKind,
};
use std::collections::VecDeque;

/// Maximum over the last `length` values of one f64 input, maintained by a descending
/// monotonic deque (front = current window max). `None` during warm-up.
pub struct RollingMax {
    length: usize,
    // descending monotonic deque of (value, stream_index); front is the window max.
    // node-owned, bounded by `length` (C7), pre-sized so push_back does not allocate.
    deque: VecDeque<(f64, u64)>,
    seen: u64,    // stream position of the next sample (drives front eviction)
    count: usize, // samples seen so far, capped at `length` — the warm-up gate
    out: [Cell; 1],
}

impl RollingMax {
    /// Build a rolling max of window `length` (must be >= 1; mirror `Sma::new`).
    pub fn new(length: usize) -> Self {
        assert!(length >= 1, "RollingMax length must be >= 1");
        Self {
            length,
            deque: VecDeque::with_capacity(length),
            seen: 0,
            count: 0,
            out: [Cell::from_f64(0.0)],
        }
    }

    /// The param-generic recipe for a blueprint primitive: declares `length` and builds
    /// through `RollingMax::new`.
    pub fn builder() -> PrimitiveBuilder {
        PrimitiveBuilder::new(
            "RollingMax",
            NodeSchema {
                inputs: vec![PortSpec { kind: ScalarKind::F64, firing: Firing::Any, name: "series".into() }],
                output: vec![FieldSpec { name: "value".into(), kind: ScalarKind::F64 }],
                params: vec![ParamSpec { name: "length".into(), kind: ScalarKind::I64 }],
            },
            |p| Box::new(RollingMax::new(p[0].i64() as usize)),
        )
    }
}

impl Node for RollingMax {
    // The window lives in node state, so only the newest sample is read each cycle.
    fn lookbacks(&self) -> Vec<usize> {
        vec![1]
    }

    fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> {
        let w = ctx.f64_in(0);
        if w.is_empty() {
            return None; // no sample yet
        }
        let x = w[0]; // index 0 = newest (financial indexing)
        let idx = self.seen;
        self.seen += 1;
        // descending invariant: a back element <= x can never again be the max — drop it.
        while let Some(&(v, _)) = self.deque.back() {
            if v <= x {
                self.deque.pop_back();
            } else {
                break;
            }
        }
        self.deque.push_back((x, idx));
        // evict the front once it leaves the window [idx-length+1, idx] (no subtraction:
        // front index `i` is out when `i + length <= idx`).
        while let Some(&(_, i)) = self.deque.front() {
            if i + self.length as u64 <= idx {
                self.deque.pop_front();
            } else {
                break;
            }
        }
        if self.count < self.length {
            self.count += 1;
        }
        if self.count < self.length {
            return None; // not yet warmed up
        }
        self.out[0] = Cell::from_f64(self.deque.front().expect("deque non-empty after push").0);
        Some(&self.out)
    }

    fn label(&self) -> String {
        format!("RollingMax({})", self.length)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use aura_core::{AnyColumn, Scalar, Timestamp};

    fn drive(node: &mut RollingMax, feed: &[f64]) -> Vec<Option<f64>> {
        let mut inputs = vec![AnyColumn::with_capacity(ScalarKind::F64, 1)];
        let mut out = Vec::new();
        for &v in feed {
            inputs[0].push(Scalar::f64(v)).unwrap();
            out.push(node.eval(Ctx::new(&inputs, Timestamp(0))).map(|r| r[0].f64()));
        }
        out
    }

    #[test]
    fn rolling_max_warms_up_then_tracks_the_window_max() {
        // max of [1,3,2], [3,2,5], [2,5,4] once warmed up; silent for the first two.
        let mut node = RollingMax::new(3);
        let got = drive(&mut node, &[1.0, 3.0, 2.0, 5.0, 4.0]);
        assert_eq!(got, vec![None, None, Some(3.0), Some(5.0), Some(5.0)]);
    }

    #[test]
    fn rolling_max_evicts_the_expiring_window_maximum() {
        // the front-eviction path: the early big value must leave the window. length 2.
        // windows: [9,1]->9, [1,2]->2, [2,3]->3 (the 9 has expired, not stuck as max).
        let mut node = RollingMax::new(2);
        let got = drive(&mut node, &[9.0, 1.0, 2.0, 3.0]);
        assert_eq!(got, vec![None, Some(9.0), Some(2.0), Some(3.0)]);
    }

    #[test]
    fn rolling_max_length_one_is_identity() {
        let mut node = RollingMax::new(1);
        assert_eq!(drive(&mut node, &[7.0, 9.0, 4.0]), vec![Some(7.0), Some(9.0), Some(4.0)]);
    }

    #[test]
    fn deque_matches_naive_window_max() {
        // the monotonic deque must equal a naive O(N) per-window max at every warmed
        // cycle, across a long deterministic series (mirrors Sma's re-sum cross-check).
        let length = 37;
        let mut node = RollingMax::new(length);
        let series: Vec<f64> =
            (0..3_000).map(|i| ((i * 1103515245 + 12345) % 1000) as f64 * 0.5 - 250.0).collect();
        let mut inputs = vec![AnyColumn::with_capacity(ScalarKind::F64, 1)];
        for (t, &x) in series.iter().enumerate() {
            inputs[0].push(Scalar::f64(x)).unwrap();
            let got = node.eval(Ctx::new(&inputs, Timestamp(0))).map(|r| r[0].f64());
            if t + 1 >= length {
                let want = series[t + 1 - length..=t].iter().cloned().fold(f64::NEG_INFINITY, f64::max);
                assert_eq!(got, Some(want), "at t={t}");
            } else {
                assert_eq!(got, None, "silent until warmed up at t={t}");
            }
        }
    }

    #[test]
    fn lookback_is_one_window_lives_in_node_state() {
        assert_eq!(RollingMax::new(20).lookbacks(), vec![1]);
    }

    #[test]
    fn length_must_be_at_least_one() {
        let r = std::panic::catch_unwind(|| RollingMax::new(0));
        assert!(r.is_err(), "RollingMax::new(0) must panic");
    }

    #[test]
    fn label_carries_the_window() {
        assert_eq!(RollingMax::new(20).label(), "RollingMax(20)");
    }
}

• Step 2: Register the module + re-export in lib.rs

In crates/aura-std/src/lib.rs, add mod rolling_max; in the alphabetical mod block (between mod resample; and mod series_reducer;), and pub use rolling_max::RollingMax; in the pub use block (near pub use sma::Sma;).

• Step 3: Run the tests to verify they pass

Run: cargo test -p aura-std rolling_max Expected: PASS — rolling_max_warms_up_then_tracks_the_window_max, rolling_max_evicts_the_expiring_window_maximum, rolling_max_length_one_is_identity, deque_matches_naive_window_max, lookback_is_one_window_lives_in_node_state, length_must_be_at_least_one, label_carries_the_window all pass (7 tests).

(The module must compile first; if Step 1's tests are written before Step 2's registration, cargo test cannot see the file — Step 2 is part of making them run.)

---

Task 2: RollingMin node

Files:

• Create: crates/aura-std/src/rolling_min.rs

• Modify: crates/aura-std/src/lib.rs:18-66

• Step 1: Write the node file with its failing tests

Create crates/aura-std/src/rolling_min.rs — the mirror of RollingMax: an ascending monotonic deque (pop the back while v >= x; front = current window min), warm-up skip-emit, one length I64 param.

//! `RollingMin` — minimum over the last `length` values of one f64 input.
//!
//! The mirror of `RollingMax`: an **ascending** monotonic deque (front = window min).
//! Each cycle pops from the back every element >= the new sample, pushes the new one,
//! and evicts the front once it leaves the window. See `rolling_max.rs` for the shared
//! rationale (O(1) amortized, window in node state, warm-up skip-emit, C7).

use aura_core::{
    Cell, Ctx, FieldSpec, Firing, Node, NodeSchema, ParamSpec, PortSpec, PrimitiveBuilder,
    ScalarKind,
};
use std::collections::VecDeque;

/// Minimum over the last `length` values of one f64 input, maintained by an ascending
/// monotonic deque (front = current window min). `None` during warm-up.
pub struct RollingMin {
    length: usize,
    deque: VecDeque<(f64, u64)>, // ascending; front = window min
    seen: u64,
    count: usize,
    out: [Cell; 1],
}

impl RollingMin {
    /// Build a rolling min of window `length` (must be >= 1).
    pub fn new(length: usize) -> Self {
        assert!(length >= 1, "RollingMin length must be >= 1");
        Self {
            length,
            deque: VecDeque::with_capacity(length),
            seen: 0,
            count: 0,
            out: [Cell::from_f64(0.0)],
        }
    }

    /// The param-generic recipe for a blueprint primitive: declares `length` and builds
    /// through `RollingMin::new`.
    pub fn builder() -> PrimitiveBuilder {
        PrimitiveBuilder::new(
            "RollingMin",
            NodeSchema {
                inputs: vec![PortSpec { kind: ScalarKind::F64, firing: Firing::Any, name: "series".into() }],
                output: vec![FieldSpec { name: "value".into(), kind: ScalarKind::F64 }],
                params: vec![ParamSpec { name: "length".into(), kind: ScalarKind::I64 }],
            },
            |p| Box::new(RollingMin::new(p[0].i64() as usize)),
        )
    }
}

impl Node for RollingMin {
    fn lookbacks(&self) -> Vec<usize> {
        vec![1]
    }

    fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> {
        let w = ctx.f64_in(0);
        if w.is_empty() {
            return None;
        }
        let x = w[0];
        let idx = self.seen;
        self.seen += 1;
        // ascending invariant: a back element >= x can never again be the min — drop it.
        while let Some(&(v, _)) = self.deque.back() {
            if v >= x {
                self.deque.pop_back();
            } else {
                break;
            }
        }
        self.deque.push_back((x, idx));
        while let Some(&(_, i)) = self.deque.front() {
            if i + self.length as u64 <= idx {
                self.deque.pop_front();
            } else {
                break;
            }
        }
        if self.count < self.length {
            self.count += 1;
        }
        if self.count < self.length {
            return None;
        }
        self.out[0] = Cell::from_f64(self.deque.front().expect("deque non-empty after push").0);
        Some(&self.out)
    }

    fn label(&self) -> String {
        format!("RollingMin({})", self.length)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use aura_core::{AnyColumn, Scalar, Timestamp};

    fn drive(node: &mut RollingMin, feed: &[f64]) -> Vec<Option<f64>> {
        let mut inputs = vec![AnyColumn::with_capacity(ScalarKind::F64, 1)];
        let mut out = Vec::new();
        for &v in feed {
            inputs[0].push(Scalar::f64(v)).unwrap();
            out.push(node.eval(Ctx::new(&inputs, Timestamp(0))).map(|r| r[0].f64()));
        }
        out
    }

    #[test]
    fn rolling_min_warms_up_then_tracks_the_window_min() {
        // min of [9,7,8], [7,8,5], [8,5,6]; silent for the first two.
        let mut node = RollingMin::new(3);
        let got = drive(&mut node, &[9.0, 7.0, 8.0, 5.0, 6.0]);
        assert_eq!(got, vec![None, None, Some(7.0), Some(5.0), Some(5.0)]);
    }

    #[test]
    fn rolling_min_evicts_the_expiring_window_minimum() {
        // length 2; the early small value must leave the window: [1,9]->1, [9,8]->8, [8,7]->7.
        let mut node = RollingMin::new(2);
        let got = drive(&mut node, &[1.0, 9.0, 8.0, 7.0]);
        assert_eq!(got, vec![None, Some(1.0), Some(8.0), Some(7.0)]);
    }

    #[test]
    fn rolling_min_length_one_is_identity() {
        let mut node = RollingMin::new(1);
        assert_eq!(drive(&mut node, &[7.0, 9.0, 4.0]), vec![Some(7.0), Some(9.0), Some(4.0)]);
    }

    #[test]
    fn deque_matches_naive_window_min() {
        let length = 37;
        let mut node = RollingMin::new(length);
        let series: Vec<f64> =
            (0..3_000).map(|i| ((i * 1103515245 + 12345) % 1000) as f64 * 0.5 - 250.0).collect();
        let mut inputs = vec![AnyColumn::with_capacity(ScalarKind::F64, 1)];
        for (t, &x) in series.iter().enumerate() {
            inputs[0].push(Scalar::f64(x)).unwrap();
            let got = node.eval(Ctx::new(&inputs, Timestamp(0))).map(|r| r[0].f64());
            if t + 1 >= length {
                let want = series[t + 1 - length..=t].iter().cloned().fold(f64::INFINITY, f64::min);
                assert_eq!(got, Some(want), "at t={t}");
            } else {
                assert_eq!(got, None, "silent until warmed up at t={t}");
            }
        }
    }

    #[test]
    fn lookback_is_one_window_lives_in_node_state() {
        assert_eq!(RollingMin::new(20).lookbacks(), vec![1]);
    }

    #[test]
    fn length_must_be_at_least_one() {
        let r = std::panic::catch_unwind(|| RollingMin::new(0));
        assert!(r.is_err(), "RollingMin::new(0) must panic");
    }

    #[test]
    fn label_carries_the_window() {
        assert_eq!(RollingMin::new(20).label(), "RollingMin(20)");
    }
}

• Step 2: Register the module + re-export in lib.rs

In crates/aura-std/src/lib.rs, add mod rolling_min; (after mod rolling_max;) and pub use rolling_min::RollingMin; (after pub use rolling_max::RollingMax;).

• Step 3: Run the tests to verify they pass

Run: cargo test -p aura-std rolling_min Expected: PASS — 7 tests (rolling_min_warms_up_then_tracks_the_window_min, rolling_min_evicts_the_expiring_window_minimum, rolling_min_length_one_is_identity, deque_matches_naive_window_min, lookback_is_one_window_lives_in_node_state, length_must_be_at_least_one, label_carries_the_window).

---

Task 3: Breakout signal composition test (C2 causality + ±1 latch)

Validates the breakout composition — close → Delay(1) → {RollingMax,RollingMin}(N) → {Gt,Gt} → {Latch,Latch} → Sub — directly via GraphBuilder/Harness, using the new nodes (Tasks 1-2) + existing Delay/Gt/Latch/Sub. Independent of the CLI binary (which cannot be imported by an integration test). Pins the two load-bearing acceptance properties: C2 (the channel excludes the current bar) and the held ±1 direction. New file name avoids the existing unrelated ger40_breakout.rs.

Files:

• Create: crates/aura-engine/tests/stage1_breakout_e2e.rs

• Step 1: Write the failing composition test

Create crates/aura-engine/tests/stage1_breakout_e2e.rs:

//! Breakout signal composition: close -> Delay(1) -> {RollingMax,RollingMin}(N) ->
//! {Gt,Gt} -> {Latch,Latch} -> Sub = bias in {-1,0,+1}. Tests the C2 causality guard
//! (the channel excludes the current bar via Delay(1)) and the held ±1 direction,
//! built straight from aura-std nodes (no CLI dependency).

use aura_core::{Scalar, Timestamp};
use aura_engine::{GraphBuilder, Harness, Source, VecSource};
use aura_std::{Delay, Gt, Latch, Recorder, RollingMax, RollingMin, Sub};
use std::sync::mpsc;

// A minimal single-f64 source role "price" fed into the breakout signal subgraph,
// tapping the Sub (bias) output through a Recorder. channel N = 3.
fn run_breakout_bias(closes: &[f64]) -> Vec<(Timestamp, f64)> {
    let (tx, rx) = mpsc::channel();
    let mut g = GraphBuilder::new("breakout_sig");
    let delay = g.add(Delay::builder().bind("lag", Scalar::i64(1)));
    let mx = g.add(RollingMax::builder().bind("length", Scalar::i64(3)));
    let mn = g.add(RollingMin::builder().bind("length", Scalar::i64(3)));
    let gt_up = g.add(Gt::builder());
    let gt_down = g.add(Gt::builder());
    let up_latch = g.add(Latch::builder());
    let down_latch = g.add(Latch::builder());
    let bias = g.add(Sub::builder());
    let rec = g.add(Recorder::builder(vec![aura_core::ScalarKind::F64], aura_core::Firing::Any, tx));
    let price = g.source_role("price", aura_core::ScalarKind::F64);
    g.feed(price, [delay.input("series"), gt_up.input("a"), gt_down.input("b")]);
    g.connect(delay.output("value"), mx.input("series"));
    g.connect(delay.output("value"), mn.input("series"));
    g.connect(mx.output("value"), gt_up.input("b")); // close > max(prior N)
    g.connect(mn.output("value"), gt_down.input("a")); // min(prior N) > close
    g.connect(gt_up.output("value"), up_latch.input("set"));
    g.connect(gt_down.output("value"), up_latch.input("reset"));
    g.connect(gt_down.output("value"), down_latch.input("set"));
    g.connect(gt_up.output("value"), down_latch.input("reset"));
    g.connect(up_latch.output("value"), bias.input("lhs"));
    g.connect(down_latch.output("value"), bias.input("rhs"));
    g.connect(bias.output("value"), rec.input("col[0]"));
    let flat = g.build().expect("breakout signal wiring resolves").compile_with_params(&[]).expect("compiles");
    let mut h = Harness::bootstrap(flat).expect("bootstraps");
    let prices: Vec<(Timestamp, f64)> = closes.iter().enumerate().map(|(i, &c)| (Timestamp(i as i64), c)).collect();
    let src: Vec<Box<dyn Source>> = vec![Box::new(VecSource::new(prices))];
    h.run(src);
    rx.try_iter().map(|(ts, row): (Timestamp, Vec<Scalar>)| (ts, row[0].as_f64())).collect()
}

#[test]
fn breakout_is_causal_and_holds_pm1_direction() {
    // closes: an up-break at bar 3, quiet hold, a down-break at bar 6.
    //   idx:    0    1    2    3    4    5   6    7
    let closes = [100.0, 101.0, 99.0, 102.0, 100.0, 100.0, 95.0, 100.0];
    // bar3: max(close[0..2])=101, close[3]=102 > 101 -> UP-break. (C2: a window INCLUDING
    //       bar3 would be max=102, and 102>102 strict is false -> no break; the break
    //       firing here proves the current bar is excluded.)
    // bars4,5: no break -> bias HOLDS +1.
    // bar6: min(close[3..5])=100, close[6]=95 < 100 -> DOWN-break -> bias flips to -1.
    // bar7: no break -> HOLDS -1.
    let got = run_breakout_bias(&closes);
    let vals: Vec<f64> = got.iter().map(|(_, v)| *v).collect();
    // bias is emitted only once the channel warms up (Delay(1)+RollingMax(3) -> first at bar 3).
    assert_eq!(vals, vec![1.0, 1.0, 1.0, -1.0, -1.0], "expected +1 held then -1 held; got {vals:?}");
}

• Step 2: Run the test to verify it passes

Run: cargo test -p aura-engine --test stage1_breakout_e2e Expected: PASS — breakout_is_causal_and_holds_pm1_direction (1 test). The GraphBuilder/source_role/feed/connect/build + compile_with_params(&[]) + Harness::bootstrap + h.run(Vec<Box<dyn Source>>) idiom is the one stage1_r_graph / run_stage1_r use in crates/aura-cli/src/main.rs (NOT the raw FlatGraph/Edge/Target wiring in ger40_breakout.rs); if a name differs, match main.rs — the asserted bias sequence [+1,+1,+1,-1,-1] is the fixed contract.

---

Task 4: CLI breakout strategy (stage1_breakout_graph, family, --strategy/--channel)

All main.rs changes land together: the Strategy variant, its parse + dispatch arms (the dispatch match is exhaustive — rustc forces the arm, so the family it calls must exist in the same task), the graph builder, the family, and the --channel flag

• grid field. Ends with a green workspace build + the CLI seam test.

Files:

• Modify: crates/aura-cli/src/main.rs (imports; enum 1286; parse 1336; dispatch 1422; grid 1123-1140 + 1345; new fns)

• Test: crates/aura-cli/tests/cli_run.rs

• Step 1: Extend imports

In crates/aura-cli/src/main.rs, add Delay, Gt, Latch, RollingMax, RollingMin to the use aura_std::{...} block (~lines 30-33; Sub is already there, but Delay/Gt/Latch are not — stage1_r_graph never used them), and add SweepPoint to the use aura_engine::{...} block (~lines 19-23; SweepFamily is already there, SweepPoint is not).

• Step 2: Add the channel grid field

In crates/aura-cli/src/main.rs:1123-1140, add a channel: Vec<i64> field to struct Stage1RGrid and channel: vec![1920] to its impl Default:

#[derive(Clone, Debug, PartialEq)]
struct Stage1RGrid {
    fast: Vec<i64>,
    slow: Vec<i64>,
    stop_length: Vec<i64>,
    stop_k: Vec<f64>,
    channel: Vec<i64>,
}

impl Default for Stage1RGrid {
    fn default() -> Self {
        Self {
            fast: vec![2, 3],
            slow: vec![6, 12],
            stop_length: vec![STAGE1_R_STOP_LENGTH],
            stop_k: vec![STAGE1_R_STOP_K],
            channel: vec![1920],
        }
    }
}

(The two defaults tests at main.rs:3113 and :3154 compare against Stage1RGrid::default() / a .clone(), not a struct literal, so they stay green with the new field. The stage1-r family never reads channel, so its grid + goldens are untouched.)

• Step 3: Add the --channel parse arm

In crates/aura-cli/src/main.rs:1345-1348, add to the flag match (alongside --fast/--slow/--stop-length/--stop-k):

            "--channel" => grid.channel = parse_csv_list(value).map_err(|()| usage())?,

• Step 4: Add the Strategy::Stage1Breakout variant + parse + dispatch arms

In crates/aura-cli/src/main.rs:

enum Strategy (1286-1290):

enum Strategy {
    SmaCross,
    Momentum,
    Stage1R,
    Stage1Breakout,
}

parse arm (after the "stage1-r" arm at 1339):

                    "stage1-breakout" => Strategy::Stage1Breakout,

dispatch arm (after the Strategy::Stage1R arm at 1425):

        Strategy::Stage1Breakout => stage1_breakout_sweep_family(persist.then_some(name), &data, grid),

• Step 5: Add stage1_breakout_graph

In crates/aura-cli/src/main.rs, near stage1_r_graph, add (the broker / executor / taps / reduce branch are copied verbatim from stage1_r_graph 1968-2026; only the signal leg and the bound vol-stop differ):

/// The stage1-r harness with its signal leg swapped for a Donchian channel breakout:
/// `close -> Delay(1) -> {RollingMax,RollingMin}(channel) -> {Gt,Gt} -> {Latch,Latch}
/// -> Sub = bias in {-1,0,+1}`. The one Delay(1) on close feeds both rolling nodes, so
/// each channel covers `close[t-N..t-1]` (the C2 guard: the current bar is excluded).
/// `channel = None` leaves the lengths open (unused today; the family binds them); the
/// stop is bound (defines R), identical downstream to stage1_r_graph.
fn stage1_breakout_graph(
    tx_eq: mpsc::Sender<(Timestamp, Vec<Scalar>)>,
    tx_ex: mpsc::Sender<(Timestamp, Vec<Scalar>)>,
    tx_r: mpsc::Sender<(Timestamp, Vec<Scalar>)>,
    tx_req: mpsc::Sender<(Timestamp, Vec<Scalar>)>,
    channel: Option<i64>,
    stop_length: i64,
    stop_k: f64,
    reduce: bool,
) -> Composite {
    let mut g = GraphBuilder::new("stage1_breakout");
    // Donchian breakout signal leg.
    let delay = g.add(Delay::builder().bind("lag", Scalar::i64(1)));
    let mut mx_b = RollingMax::builder().named("channel_hi");
    let mut mn_b = RollingMin::builder().named("channel_lo");
    if let Some(n) = channel {
        mx_b = mx_b.bind("length", Scalar::i64(n));
        mn_b = mn_b.bind("length", Scalar::i64(n));
    }
    let mx = g.add(mx_b);
    let mn = g.add(mn_b);
    let gt_up = g.add(Gt::builder());
    let gt_down = g.add(Gt::builder());
    let up_latch = g.add(Latch::builder());
    let down_latch = g.add(Latch::builder());
    let exposure = g.add(Sub::builder()); // up_latch - down_latch -> bias in {-1,0,+1}
    // pip branch (verbatim from stage1_r_graph).
    let broker = g.add(SimBroker::builder(SYNTHETIC_PIP_SIZE));
    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(risk_executor(StopRule::Vol { length: stop_length, k: 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,
        [
            delay.input("series"),
            gt_up.input("a"),
            gt_down.input("b"),
            broker.input("price"),
            exec.input("price"),
        ],
    );
    g.connect(delay.output("value"), mx.input("series"));
    g.connect(delay.output("value"), mn.input("series"));
    g.connect(mx.output("value"), gt_up.input("b"));
    g.connect(mn.output("value"), gt_down.input("a"));
    g.connect(gt_up.output("value"), up_latch.input("set"));
    g.connect(gt_down.output("value"), up_latch.input("reset"));
    g.connect(gt_down.output("value"), down_latch.input("set"));
    g.connect(gt_up.output("value"), down_latch.input("reset"));
    g.connect(up_latch.output("value"), exposure.input("lhs"));
    g.connect(down_latch.output("value"), exposure.input("rhs"));
    g.connect(exposure.output("value"), broker.input("exposure"));
    g.connect(exposure.output("value"), ex.input("col[0]"));
    g.connect(exposure.output("value"), 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_breakout wiring resolves")
}

• Step 6: Add stage1_breakout_sweep_family

In crates/aura-cli/src/main.rs, near stage1_r_sweep_family, add (the metrics reduce/trace branch is copied verbatim from stage1_r_sweep_family 1232-1258):

/// `aura sweep --strategy stage1-breakout`: sweep the breakout harness over a channel ×
/// stop grid. One `channel` length drives BOTH rolling nodes (parameter-ganging, #61),
/// so the family iterates the cartesian product MANUALLY with a fully-bound graph per
/// point (compile_with_params(&[]) + Harness::bootstrap, like run_stage1_r) rather than
/// the open .axis/bootstrap_with_cells path. Each member folds the dense R-record via
/// summarize_r, so the family is rankable by sqn / expectancy_r / ... (parity with
/// stage1-r). With --trace, raw recorders persist the per-cycle streams.
fn stage1_breakout_sweep_family(trace: Option<&str>, data: &DataSource, grid: &Stage1RGrid) -> SweepFamily {
    let pip = data.pip_size();
    let window = data.full_window();
    let mut varying: HashSet<String> = HashSet::new();
    if grid.channel.len() > 1 {
        varying.insert("channel".to_string());
    }
    if grid.stop_length.len() > 1 {
        varying.insert("stop_length".to_string());
    }
    if grid.stop_k.len() > 1 {
        varying.insert("stop_k".to_string());
    }
    let mut points = Vec::new();
    for &c in &grid.channel {
        for &sl in &grid.stop_length {
            for &sk in &grid.stop_k {
                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 flat = stage1_breakout_graph(tx_eq, tx_ex, tx_r, tx_req, Some(c), sl, sk, reduce)
                    .compile_with_params(&[])
                    .expect("valid stage1-breakout blueprint");
                let mut h = Harness::bootstrap(flat).expect("valid stage1-breakout harness");
                h.run(data.run_sources());
                let named: Vec<(String, Scalar)> = vec![
                    ("channel".to_string(), Scalar::i64(c)),
                    ("stop_length".to_string(), Scalar::i64(sl)),
                    ("stop_k".to_string(), Scalar::f64(sk)),
                ];
                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 {
                    let r_rows: Vec<(Timestamp, Vec<Scalar>)> = 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 {
                    let eq_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_eq.try_iter().collect();
                    let ex_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_ex.try_iter().collect();
                    let r_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_r.try_iter().collect();
                    let req_rows: Vec<(Timestamp, Vec<Scalar>)> = 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
                };
                points.push(SweepPoint { params: vec![], report: RunReport { manifest, metrics } });
            }
        }
    }
    SweepFamily { space: vec![], points }
}

• Step 7: Run the workspace build + suite to verify green

Run: cargo build --workspace Expected: clean build, 0 errors (the exhaustive dispatch match now has its Stage1Breakout arm; the new fns compile).

Run: cargo test --workspace Expected: PASS — all existing suites still green (stage1-r / sma / momentum goldens byte-identical; the breakout adds a new path, touches no existing graph), plus the new node + composition tests from Tasks 1-3.

• Step 8: Write the CLI seam test

In crates/aura-cli/tests/cli_run.rs, add (mirrors sweep_strategy_stage1_r_folded_no_trace_metrics_equal_raw_trace_metrics at :2000; reuses the in-file BIN, temp_cwd, metrics_object, Command already used there):

/// The breakout strategy reaches the CLI seam: `aura sweep --strategy stage1-breakout`
/// emits an R-bearing member, and the folded no-trace path equals the raw --trace path
/// byte-for-byte (parity with the stage1-r fold-vs-raw guard). channel 3 warms up on
/// the synthetic stream; one channel value × the single default stop = one member.
#[test]
fn sweep_strategy_stage1_breakout_folded_no_trace_metrics_equal_raw_trace_metrics() {
    let cwd = temp_cwd("sweep-stage1-breakout-fold-vs-raw");

    let folded = Command::new(BIN)
        .args(["sweep", "--strategy", "stage1-breakout", "--channel", "3"])
        .current_dir(&cwd)
        .output()
        .expect("spawn folded (no-trace) stage1-breakout sweep");
    assert!(
        folded.status.success(),
        "folded no-trace breakout sweep exit: {:?}; stderr: {}",
        folded.status,
        String::from_utf8_lossy(&folded.stderr)
    );
    let folded_out = String::from_utf8(folded.stdout).expect("utf-8");

    let raw = Command::new(BIN)
        .args(["sweep", "--strategy", "stage1-breakout", "--channel", "3", "--trace", "b1"])
        .current_dir(&cwd)
        .output()
        .expect("spawn raw (--trace) stage1-breakout sweep");
    assert!(
        raw.status.success(),
        "raw --trace breakout sweep exit: {:?}; stderr: {}",
        raw.status,
        String::from_utf8_lossy(&raw.stderr)
    );
    let raw_out = String::from_utf8(raw.stdout).expect("utf-8");

    let folded_lines: Vec<&str> = folded_out.lines().collect();
    let raw_lines: Vec<&str> = raw_out.lines().collect();
    assert_eq!(folded_lines.len(), 1, "folded breakout sweep must print 1 member: {folded_out:?}");
    assert_eq!(
        raw_lines.len(),
        folded_lines.len(),
        "member count must match: folded {} vs raw {}",
        folded_lines.len(),
        raw_lines.len()
    );

    for (i, (f, r)) in folded_lines.iter().zip(raw_lines.iter()).enumerate() {
        let fm = metrics_object(f);
        let rm = metrics_object(r);
        assert!(fm.contains("\"r\":{"), "folded breakout member {i} must carry an r block: {fm}");
        assert_eq!(
            fm, rm,
            "folded vs raw breakout metrics diverge at member {i}\n  folded: {fm}\n  raw:    {rm}"
        );
    }
    let _ = std::fs::remove_dir_all(&cwd);
}

• Step 9: Run the seam test + final gates

Run: cargo test -p aura-cli --test cli_run stage1_breakout Expected: PASS — the new stage1-breakout seam test(s).

Run: cargo test --workspace Expected: PASS (whole workspace).

Run: cargo clippy --workspace --all-targets -- -D warnings Expected: clean, no warnings.

---