Aura/docs/plans/0068-stage1-r-cli-surface.md

Stage-1 R CLI surface (iteration 3) — Implementation Plan

Parent spec: docs/specs/0065-stage1-r-signal-quality.md (the "## Iteration 3 — the CLI surface (#129)" section).

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

Goal: make the Stage-1 R layer operable from the command line — aura run --harness stage1-r [--real <SYM>] [--trace <n>] runs a backtest and prints its signal-quality metrics (the R block) in the RunReport JSON, in one shell call.

Architecture: promote vol_stop + risk_executor (+ a StopRule axis) from test fixtures to public aura-engine composite-builders; add a stage1_r_blueprint that fans one bias into both SimBroker (pip) and the risk_executor (R); add a parse_run_args tokenizer + --harness <name> selector routing to the built-in harnesses; fold summarize_r → RunMetrics.r in the run path; extend persist_traces with a third r_equity tap for aura chart --tap r_equity.

Tech Stack: aura-engine (GraphBuilder/Composite, report), aura-std nodes (Sma/Sub/Bias/SimBroker/Recorder/LinComb/FixedStop/Sizer/PositionManagement + vol_stop primitives), aura-cli (hand-rolled argv dispatch, trace store).

Files this plan creates or modifies:

• Modify: crates/aura-engine/Cargo.toml — move aura-std from [dev-dependencies] to [dependencies].
• Create: crates/aura-engine/src/composites.rs — vol_stop, StopRule, risk_executor public builders.
• Modify: crates/aura-engine/src/lib.rs:45-67 — mod composites; + pub use composites::{...}.
• Modify: crates/aura-engine/tests/vol_stop_composite.rs — delete local fn vol_stop, import the public one.
• Modify: crates/aura-engine/tests/risk_executor.rs — delete local fn risk_executor, call risk_executor(StopRule::Fixed(d), b); add a Vol-backed RED test.
• Modify: crates/aura-cli/src/main.rs — stage1_r_blueprint, run_stage1_r, a new additive persist_traces_r 3-tap helper (the existing 2-tap persist_traces + its 7 pip callers stay byte-unchanged), parse_run_args/RunArgs/HarnessKind/RunData/run_dispatch, the run argv arm, USAGE.
• Test: crates/aura-cli/tests/cli_run.rs — stage1-r smoke, selector, flag-composition, r_equity round-trip.
• Test: crates/aura-cli/src/main.rs (inline #[cfg(test)]) — parse_run_args grammar unit tests.

Architecture decisions pinned (one sentence each, decided by the orchestrator):

• Composites live in aura-engine (not aura-std, not a new crate): they are the engine's convenience/execution layer over the standard nodes — the sibling tier of report::summarize_r, already in aura-engine — and this yields a fully acyclic normal-dependency graph (aura-engine → aura-std → aura-core) where homing them in aura-std would force a Cargo dev-dependency cycle.
• The r_equity summing LinComb lives in the stage1_r_blueprint (not inside risk_executor): the R-equity series is a charting-specific derivation of the CLI harness, so the reusable executor keeps exposing exactly the PM record.
• risk_executor's body is the fixture verbatim except the one stop g.add line, which becomes a match StopRule { Fixed => g.add(FixedStop…), Vol => g.add(vol_stop…) } — every downstream edge is arm-independent because both stops expose price → stop_distance.
• The synthetic stage1-r harness uses StopRule::Vol { length: 3, k: 2.0 } over a ~18-tick rise-fall-rise stream so the short vol stop warms up and the SMA-cross bias flips at least once (≥1 closed trade for the smoke test).

---

Task 1: Promote vol_stop + risk_executor + StopRule to public aura-engine API

Files:

• Modify: crates/aura-engine/Cargo.toml

• Create: crates/aura-engine/src/composites.rs

• Modify: crates/aura-engine/src/lib.rs:45-67

• Modify: crates/aura-engine/tests/vol_stop_composite.rs

• Modify: crates/aura-engine/tests/risk_executor.rs

• Step 1: Move aura-std to a normal dependency

In crates/aura-engine/Cargo.toml, add to the [dependencies] section (after the serde_json line):

# aura-std's standard nodes back the public composite-builders in src/composites.rs
# (vol_stop, risk_executor) — the engine's convenience layer over the standard nodes.
# aura-std depends only on aura-core, so this stays acyclic (aura-engine -> aura-std
# -> aura-core).
aura-std = { path = "../aura-std" }

and DELETE the now-redundant line under [dev-dependencies]:

aura-std = { path = "../aura-std" }

(The [dev-dependencies] section keeps chrono / chrono-tz.)

• Step 2: Create crates/aura-engine/src/composites.rs

//! Reusable Stage-1 composite-builders: the volatility stop and the per-symbol
//! RiskExecutor, promoted from the iter-2 integration-test fixtures so the CLI
//! harness (and any consumer) can wire them. Both are `GraphBuilder` compositions
//! of `aura-std` primitives — the engine's convenience layer over the standard
//! nodes (the sibling of `report::summarize_r`, which likewise lives in this crate
//! and reads the `aura-std` PositionManagement record). The Veto is a documented
//! seam, not a node (a pass-through identity is exactly what C19/C23 DCE deletes),
//! so it appears nowhere here.
use aura_core::Scalar;
use aura_std::{
    Delay, Ema, FixedStop, LinComb, Mul, PositionManagement, Sizer, Sqrt, Sub, PM_FIELD_NAMES,
};

use crate::{Composite, GraphBuilder};

/// The volatility stop as a composition of primitives:
/// `stop_distance = k · Sqrt(Ema(Mul(Δ,Δ), length))`, `Δ = price − Delay(price, 1)`
/// — a rolling EWMA standard deviation. Role: input `price` → output `stop_distance`.
pub fn vol_stop(length: i64, k: f64) -> Composite {
    let mut g = GraphBuilder::new("vol_stop");
    let price = g.input_role("price");
    let delay = g.add(Delay::builder().bind("lag", Scalar::i64(1))); // z^-1: prev price
    let sub = g.add(Sub::builder()); // Δ = price − prev
    let sq = g.add(Mul::builder()); // Δ·Δ = Δ²
    let ema = g.add(Ema::builder().bind("length", Scalar::i64(length))); // EWMA variance
    let sqrt = g.add(Sqrt::builder()); // → σ (price units)
    let scale = g.add(LinComb::builder(1).bind("weights[0]", Scalar::f64(k))); // k·σ
    g.feed(price, [delay.input("series"), sub.input("lhs")]);
    g.connect(delay.output("value"), sub.input("rhs"));
    g.connect(sub.output("value"), sq.input("lhs"));
    g.connect(sub.output("value"), sq.input("rhs")); // square: feed Δ to both legs
    g.connect(sq.output("value"), ema.input("series"));
    g.connect(ema.output("value"), sqrt.input("value"));
    g.connect(sqrt.output("value"), scale.input("term[0]"));
    g.expose(scale.output("value"), "stop_distance");
    g.build().expect("vol_stop composite wires")
}

/// The protective stop-rule axis (C11 structural): a fixed distance, or the
/// volatility-scaled `vol_stop`. Both expose `price → stop_distance`, so the
/// RiskExecutor embeds either by identical downstream wiring.
pub enum StopRule {
    Fixed(f64),
    Vol { length: i64, k: f64 },
}

/// The per-symbol RiskExecutor: open input roles `bias` + `price`, internal
/// `stop-rule → Sizer → PositionManagement`, exposing every field of PM's dense
/// R-record. Price fans to BOTH the stop-rule and PM; bias fans to the Sizer and PM;
/// the Sizer's `size` feeds PM's size slot. The embedded stop is the chosen `StopRule`.
pub fn risk_executor(stop: StopRule, risk_budget: f64) -> Composite {
    let mut g = GraphBuilder::new("risk_executor");
    let bias = g.input_role("bias");
    let price = g.input_role("price");
    let stop = match stop {
        StopRule::Fixed(d) => g.add(FixedStop::builder().bind("distance", Scalar::f64(d))),
        StopRule::Vol { length, k } => g.add(vol_stop(length, k)),
    };
    let sizer = g.add(Sizer::builder().bind("risk_budget", Scalar::f64(risk_budget)));
    let pm = g.add(PositionManagement::builder());
    g.feed(price, [stop.input("price"), pm.input("price")]); // price fans to stop + PM
    g.feed(bias, [sizer.input("bias"), pm.input("bias")]); // bias fans to sizer + PM
    g.connect(stop.output("stop_distance"), sizer.input("stop_distance"));
    g.connect(stop.output("stop_distance"), pm.input("stop_distance"));
    g.connect(sizer.output("size"), pm.input("size")); // flat-1R size into PM
    for field in PM_FIELD_NAMES {
        g.expose(pm.output(field), field);
    }
    g.build().expect("risk_executor wires")
}

• Step 3: Re-export from crates/aura-engine/src/lib.rs

Add mod composites; to the module block (alphabetical, after mod builder; at line 46):

mod builder;
mod composites;
mod graph_model;

Add the public re-export after the builder re-export (line 58):

pub use builder::{BuildError, GraphBuilder, InPort, NodeHandle, OutPort, RoleHandle};
pub use composites::{risk_executor, vol_stop, StopRule};

• Step 4: Verify the engine builds with the promoted API

Run: cargo build -p aura-engine Expected: builds clean (composites.rs compiles against the new normal aura-std dep).

• Step 5: Rewire crates/aura-engine/tests/vol_stop_composite.rs

Delete the local fn vol_stop(length: i64, k: f64) -> Composite { … } (lines 13-31). Replace the imports at lines 6-9 with (pruning the now-unused composite-builder deps; the compiler under -D warnings enumerates any leftover):

use aura_core::{Firing, Scalar, ScalarKind, Timestamp};
use aura_engine::{vol_stop, GraphBuilder, VecSource};
use aura_std::Recorder;
use std::sync::mpsc::channel;

The test body (vol_stop_emits_k_times_rolling_stddev, which calls g.add(vol_stop(3, 2.0)) at line 40) is unchanged — it now calls the promoted public vol_stop.

• Step 6: Rewire crates/aura-engine/tests/risk_executor.rs + add a Vol-backed RED test

Delete the local fn risk_executor(stop_distance: f64, risk_budget: f64) -> Composite { … } (lines 26-42). Update the imports at line 11-12 to import the promoted fn + StopRule and drop the now-unused composite-builder deps (compiler-pruned under -D warnings):

use aura_engine::{risk_executor, summarize_r, GraphBuilder, RunMetrics, StopRule, VecSource};
use aura_std::{PositionManagement, Recorder, PM_FIELD_NAMES, PM_RECORD_KINDS};

In run_executor (line 82), change the call at line 87 from g.add(risk_executor(stop_distance, risk_budget)) to:

    let exec = g.add(risk_executor(StopRule::Fixed(stop_distance), risk_budget));

Then ADD this RED test (new coverage: the Vol-backed executor, the match arm that did not exist before the promotion):

/// Property: the RiskExecutor embeds the `vol_stop` composite (the new `StopRule::Vol`
/// arm) and folds to a finite RMetric — the volatility-defined default the `stage1-r`
/// harness uses. A short k·σ stop over a rising-then-falling path opens a trade and
/// (stop or window-end) closes at least one, so the fold is non-empty and sane.
#[test]
fn risk_executor_vol_stop_arm_bootstraps_and_folds() {
    let (tx, rx) = channel();
    let mut g = GraphBuilder::new("vol_harness");
    let price = g.source_role("price", ScalarKind::F64);
    let strat = g.add(ConstLongBias::builder());
    let exec = g.add(risk_executor(StopRule::Vol { length: 3, k: 2.0 }, 1.0));
    let rec = g.add(Recorder::builder(PM_RECORD_KINDS.to_vec(), Firing::Any, tx));
    g.feed(price, [strat.input("price"), exec.input("price")]);
    g.connect(strat.output("bias"), exec.input("bias"));
    for (i, field) in PM_FIELD_NAMES.iter().enumerate() {
        let col: &'static str = format!("col[{i}]").leak();
        g.connect(exec.output(field), rec.input(col));
    }
    let mut h = g
        .build()
        .expect("vol_harness wires")
        .bootstrap_with_params(vec![])
        .expect("bootstraps");
    let path = [100.0, 101.0, 100.0, 101.0, 103.0, 106.0, 104.0, 99.0, 95.0, 96.0];
    let stream: Vec<(Timestamp, Scalar)> =
        path.iter().enumerate().map(|(i, &p)| (Timestamp(i as i64), Scalar::f64(p))).collect();
    h.run(vec![Box::new(VecSource::new(stream))]);
    let ledger: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
    let m = summarize_r(&ledger, 0.0);
    assert!(m.n_trades >= 1, "the vol-stop executor must fold at least one trade");
    assert!(m.sqn.is_finite(), "SQN must be finite, got {}", m.sqn);
}

(ConstLongBias stays the local producer defined at lines 48-78; Scalar, ScalarKind, Firing, Timestamp, channel are already imported by the file.)

• Step 7: Run the engine tests — fixtures green + the new Vol arm

Run: cargo test -p aura-engine risk_executor Expected: PASS — risk_executor_bootstraps_and_folds_to_expected_rmetric, risk_executor_r_invariant_under_risk_budget, folded_rmetrics_survives_runmetrics_serde_round_trip, and the new risk_executor_vol_stop_arm_bootstraps_and_folds all pass.

Run: cargo test -p aura-engine vol_stop Expected: PASS — vol_stop_emits_k_times_rolling_stddev (now calling the promoted fn).

---

Task 2: The stage1_r_blueprint harness + the run_stage1_r R fold (aura-cli)

Files:

• Modify: crates/aura-cli/src/main.rs (add stage1_r_blueprint, stage1_r_prices, run_stage1_r; extend the use of aura_engine/aura_std)

• Step 1: Write the failing smoke test (inline #[cfg(test)] mod tests in main.rs)

#[test]
fn run_stage1_r_synthetic_folds_an_r_block() {
    // the stage1-r harness scores the SMA-cross signal in R: one shell-callable run
    // yields a RunReport whose metrics.r is Some with a finite SQN and >= 1 trade.
    let report = run_stage1_r(RunData::Synthetic, None);
    let r = report.metrics.r.as_ref().expect("stage1-r run must populate metrics.r");
    assert!(r.n_trades >= 1, "expected >= 1 trade, got {}", r.n_trades);
    assert!(r.sqn.is_finite(), "SQN must be finite, got {}", r.sqn);
    assert!(r.expectancy_r.is_finite(), "E[R] must be finite, got {}", r.expectancy_r);
}

• Step 2: Run the test to verify it fails

Run: cargo test -p aura-cli run_stage1_r_synthetic_folds_an_r_block Expected: FAIL — run_stage1_r / RunData do not exist yet (compile error).

• Step 3: Add the harness, the synthetic stream, and the run handler

Ensure main.rs's imports bring in the promoted composites + the nodes the harness needs (extend the existing use aura_engine::{…} and use aura_std::{…} lines): aura_engine::{risk_executor, vol_stop, StopRule, GraphBuilder, Composite, summarize_r} (several already imported) and aura_std::{Sma, Sub, Bias, SimBroker, Recorder, LinComb, PM_FIELD_NAMES, PM_RECORD_KINDS} (most already imported — add the missing ones; the compiler enumerates).

Add the data selector enum near the other CLI types:

/// Which data a `run` drives a harness on: the built-in synthetic stream, or real M1
/// close bars for a vetted symbol over an optional window.
enum RunData {
    Synthetic,
    Real { symbol: String, from: Option<i64>, to: Option<i64> },
}

Add the synthetic stream (a rise-fall-rise path so the SMA-cross bias flips and the short vol stop warms up — reuses the macd_prices shape, kept local so a future macd_prices edit cannot silently change this harness's trades):

/// A rise-fall-rise synthetic stream for the stage1-r smoke run: long enough to warm
/// the `vol_stop(length=3)` and flip the SMA(2)/SMA(4) cross at least once, so the
/// RiskExecutor opens and closes at least one trade. Deterministic (C1).
fn stage1_r_prices() -> Vec<(Timestamp, Scalar)> {
    [
        1.0000_f64, 1.0008, 1.0021, 1.0039, 1.0062, 1.0090, 1.0083, 1.0061, 1.0034,
        1.0012, 0.9998, 1.0006, 1.0024, 1.0047, 1.0069, 1.0086, 1.0097, 1.0092,
    ]
    .iter()
    .enumerate()
    .map(|(i, &p)| (Timestamp(i as i64 + 1), Scalar::f64(p)))
    .collect()
}

Add the dual-tap harness blueprint:

/// The Stage-1 R harness: the SMA-cross → `Bias` signal fanned into BOTH a `SimBroker`
/// (pip equity, the existing pip yardstick) AND the `risk_executor` (the dense R-record,
/// the new R yardstick), so one run scores the same signal in pips and in R. Four taps:
/// equity (SimBroker), exposure (Bias), the 14-column R-record (drained into
/// `summarize_r`), and `r_equity = cum_realized_r + unrealized_r` (a single series for
/// `aura chart --tap r_equity`). All params bound at build → compiles with an empty point.
#[allow(clippy::type_complexity)]
fn stage1_r_blueprint(
    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>)>,
) -> Composite {
    let mut g = GraphBuilder::new("stage1_r");
    // SMA-cross signal → Bias (the same signal the pip sample uses).
    let fast = g.add(Sma::builder().named("fast").bind("length", Scalar::i64(2)));
    let slow = g.add(Sma::builder().named("slow").bind("length", Scalar::i64(4)));
    let spread = g.add(Sub::builder());
    let exposure = g.add(Bias::builder().named("exposure").bind("scale", Scalar::f64(0.5)));
    // pip branch (verbatim from sample_blueprint_with_sinks).
    let broker = g.add(SimBroker::builder(SYNTHETIC_PIP_SIZE));
    let eq = g.add(Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx_eq));
    let ex = g.add(Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx_ex));
    // R branch: bias + price → RiskExecutor(vol_stop) → dense R-record.
    let exec = g.add(risk_executor(StopRule::Vol { length: 3, k: 2.0 }, 1.0));
    let rrec = g.add(Recorder::builder(PM_RECORD_KINDS.to_vec(), Firing::Any, tx_r));
    // r_equity = cum_realized_r + unrealized_r — one tapped series for charting.
    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));
    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"));
    // bias fans to: broker (pip), exposure sink, and the RiskExecutor (R).
    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]"));
    // tap the dense R-record (all PM fields) for summarize_r.
    for (i, field) in PM_FIELD_NAMES.iter().enumerate() {
        let col: &'static str = format!("col[{i}]").leak();
        g.connect(exec.output(field), rrec.input(col));
    }
    // r_equity sum + tap.
    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")
}

Add the run handler (synthetic + real in one fn, mirroring run_sample / run_sample_real for the source and folding both yardsticks):

/// `aura run --harness stage1-r [--real <SYM> [--from][--to]] [--trace <n>]`: build the
/// dual-tap stage1-r harness, run it on synthetic or real M1 data, fold the pip taps via
/// `summarize` and the dense R-record via `summarize_r`, and attach the R block as
/// `RunMetrics.r = Some(..)`. Pure/deterministic (C1). round_trip_cost is 0.0 (Stage-1 is
/// frictionless; a --cost knob is a follow-on).
fn run_stage1_r(data: RunData, trace: Option<&str>) -> RunReport {
    if let Some(n) = trace
        && let Err(e) = TraceStore::open("runs").ensure_name_free(n, WriteKind::Run)
    {
        eprintln!("aura: {e}");
        std::process::exit(2);
    }
    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 flat = stage1_r_blueprint(tx_eq, tx_ex, tx_r, tx_req)
        .compile_with_params(&[])
        .expect("valid stage1-r blueprint");
    let mut h = Harness::bootstrap(flat).expect("valid stage1-r harness");

    let (sources, window, pip_size): (Vec<Box<dyn aura_engine::Source>>, _, f64) = match &data {
        RunData::Synthetic => {
            let sources: Vec<Box<dyn aura_engine::Source>> =
                vec![Box::new(VecSource::new(stage1_r_prices()))];
            let window = window_of(&sources).expect("non-empty synthetic stream");
            (sources, window, SYNTHETIC_PIP_SIZE)
        }
        RunData::Real { symbol, from, to } => {
            let spec = instrument_spec_or_refuse(symbol);
            let server =
                std::sync::Arc::new(data_server::DataServer::new(data_server::DEFAULT_DATA_PATH));
            if !server.has_symbol(symbol) {
                no_real_data(symbol);
            }
            let window = probe_window(&server, symbol, *from, *to);
            let source: Box<dyn aura_engine::Source> = match aura_ingest::M1FieldSource::open(
                &server, symbol, *from, *to, aura_ingest::M1Field::Close,
            ) {
                Some(s) => Box::new(s),
                None => no_real_data(symbol),
            };
            (vec![source], window, spec.pip_size)
        }
    };
    h.run(sources);

    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();

    let manifest = sim_optimal_manifest(
        vec![
            ("sma_fast".to_string(), Scalar::i64(2)),
            ("sma_slow".to_string(), Scalar::i64(4)),
            ("exposure_scale".to_string(), Scalar::f64(0.5)),
            ("stop".to_string(), Scalar::i64(3)), // vol_stop length
        ],
        window,
        0,
        pip_size,
    );
    if let Some(name) = trace {
        persist_traces_r(name, &manifest, &eq_rows, &ex_rows, &req_rows);
    }
    let mut metrics = summarize(&f64_field(&eq_rows, 0), &f64_field(&ex_rows, 0));
    metrics.r = Some(summarize_r(&r_rows, 0.0));
    RunReport { manifest, metrics }
}

Also add the three-tap persister run_stage1_r calls — a SEPARATE helper, NOT a fourth arg on the two-tap persist_traces: that keeps the seven pip callers (:561, :613, :1024, :1101, :1370, :1459, :1695) byte-unchanged, so the cli_run.rs "taps":["equity","exposure"] pin stays green. reduce_for_tap already decimates r_equity by its MinMax default (an equity-like curve), so the chart viewer is unchanged.

/// Persist a stage1-r run's three taps: equity (off the SimBroker), exposure (off the
/// Bias), and r_equity = cum_realized_r + unrealized_r (off the RiskExecutor). Separate
/// from the two-tap `persist_traces` so the pip handlers stay byte-unchanged on disk.
fn persist_traces_r(
    name: &str,
    manifest: &RunManifest,
    eq_rows: &[(Timestamp, Vec<Scalar>)],
    ex_rows: &[(Timestamp, Vec<Scalar>)],
    req_rows: &[(Timestamp, Vec<Scalar>)],
) {
    let taps = vec![
        ColumnarTrace::from_rows("equity", &[ScalarKind::F64], eq_rows),
        ColumnarTrace::from_rows("exposure", &[ScalarKind::F64], ex_rows),
        ColumnarTrace::from_rows("r_equity", &[ScalarKind::F64], req_rows),
    ];
    if let Err(e) = TraceStore::open("runs").write(name, manifest, &taps) {
        eprintln!("aura: trace persist failed: {e}");
        std::process::exit(2);
    }
}

(The smoke test in Step 1 does not pass --trace, so persist_traces_r is compiled-but-not-run here; the binary-driven r_equity round-trip in Task 3 exercises it once --harness stage1-r is routable.)

• Step 4: Run the test to verify it passes

Run: cargo test -p aura-cli run_stage1_r_synthetic_folds_an_r_block Expected: PASS — metrics.r is Some with n_trades >= 1 and a finite SQN. (If the synthetic stream yields 0 closed trades, lengthen/steepen stage1_r_prices until ≥1 — the test pins n_trades >= 1.)

---

Task 3: parse_run_args tokenizer + --harness selector + run_dispatch

Files:

• Modify: crates/aura-cli/src/main.rs (add RunArgs/HarnessKind/parse_run_args/run_dispatch; replace the run argv arms; update USAGE)

• Test: crates/aura-cli/tests/cli_run.rs (selector + flag-composition integration tests)

• Test: crates/aura-cli/src/main.rs inline (parse_run_args grammar unit tests)

• Step 1: Write the failing tests

Inline parser unit tests (in main.rs #[cfg(test)] mod tests):

#[test]
fn parse_run_args_defaults_to_sma_synthetic() {
    let a = parse_run_args(&[]).expect("bare run");
    assert!(matches!(a.harness, HarnessKind::Sma));
    assert!(matches!(a.data, RunData::Synthetic));
    assert!(a.trace.is_none());
}

#[test]
fn parse_run_args_composes_harness_real_and_trace_in_any_order() {
    let a = parse_run_args(&["--trace", "q1", "--harness", "stage1-r", "--real", "GER40", "--from", "100"])
        .expect("composed flags");
    assert!(matches!(a.harness, HarnessKind::Stage1R));
    match a.data {
        RunData::Real { symbol, from, to } => {
            assert_eq!(symbol, "GER40");
            assert_eq!(from, Some(100));
            assert_eq!(to, None);
        }
        _ => panic!("expected real data"),
    }
    assert_eq!(a.trace.as_deref(), Some("q1"));
}

#[test]
fn parse_run_args_macd_is_an_alias_and_rejects_harness_conflict() {
    assert!(matches!(parse_run_args(&["--macd"]).unwrap().harness, HarnessKind::Macd));
    assert!(parse_run_args(&["--macd", "--harness", "sma"]).is_err());
    assert!(parse_run_args(&["--harness", "nope"]).is_err());
    assert!(parse_run_args(&["--from", "1"]).is_err()); // --from without --real
}

cli_run integration tests (in crates/aura-cli/tests/cli_run.rs, driving the binary via Command::new(BIN) exactly like the existing tests):

#[test]
fn run_harness_stage1_r_prints_an_r_block() {
    let out = std::process::Command::new(BIN).args(["run", "--harness", "stage1-r"]).output().unwrap();
    assert!(out.status.success());
    let s = String::from_utf8(out.stdout).unwrap();
    assert!(s.contains("\"r\":{"), "stage1-r run must carry an r block: {s}");
    assert!(s.contains("\"sqn\""), "the r block must carry sqn: {s}");
}

#[test]
fn run_harness_sma_is_pip_only_no_r_block() {
    let out = std::process::Command::new(BIN).args(["run", "--harness", "sma"]).output().unwrap();
    assert!(out.status.success());
    let s = String::from_utf8(out.stdout).unwrap();
    assert!(!s.contains("\"r\":"), "a pip run must omit the r key: {s}");
}

#[test]
fn run_unknown_harness_exits_two() {
    let out = std::process::Command::new(BIN).args(["run", "--harness", "nope"]).output().unwrap();
    assert_eq!(out.status.code(), Some(2));
}

#[test]
fn stage1_r_trace_persists_r_equity_and_charts_it() {
    // `temp_cwd` (cli_run.rs:13) gives a unique CWD with no external tempfile dep — the
    // pattern the existing trace tests use; it returns a `PathBuf`, so `.join` directly.
    let dir = temp_cwd("stage1-r-trace");
    let run = Command::new(BIN)
        .current_dir(&dir)
        .args(["run", "--harness", "stage1-r", "--trace", "q1"])
        .output()
        .unwrap();
    assert!(run.status.success());
    // the third tap is persisted beside equity/exposure
    assert!(dir.join("runs/traces/q1/r_equity.json").exists(), "r_equity tap must persist");
    let chart = Command::new(BIN)
        .current_dir(&dir)
        .args(["chart", "q1", "--tap", "r_equity"])
        .output()
        .unwrap();
    assert!(chart.status.success());
    let html = String::from_utf8(chart.stdout).unwrap();
    assert!(!html.is_empty() && html.contains("r_equity"), "chart must render the r_equity series");
}

Uses temp_cwd("stage1-r-trace") (cli_run.rs:13) — the existing trace tests' pattern, no external tempfile dep — and Command (imported at cli_run.rs:4). The plain-run tap-order pin run_trace_index_carries_manifest_and_ordered_tap_list (cli_run.rs:329) must stay green: only the stage1-r harness writes a third tap. If the html.contains("r_equity") label differs, mirror the assertion shape of the existing chart_emits_self_contained_html_for_a_persisted_run (cli_run.rs:416).

• Step 2: Run the tests to verify they fail

Run: cargo test -p aura-cli parse_run_args Expected: FAIL — parse_run_args / RunArgs / HarnessKind do not exist (compile error).

• Step 3: Add the grammar types, the parser, and the dispatcher

/// Which built-in harness `aura run` drives. A fixed compile-time enumeration over
/// Rust-authored harnesses — NOT a runtime node registry and NOT a DSL (C9/C17): the
/// CLI *runs* an authored harness, it does not wire one.
enum HarnessKind {
    Sma,
    Macd,
    Stage1R,
}

/// The parsed `aura run` invocation: a harness, a data source, and an optional trace name.
struct RunArgs {
    harness: HarnessKind,
    data: RunData,
    trace: Option<String>,
}

/// Parse the `run` tail: `[--harness <name>] [--macd] [--real <SYM> [--from <ms>] [--to <ms>]]
/// [--trace <name>]` in any order, each flag at most once. `--harness sma` is the default;
/// `--macd` is a back-compat alias for `--harness macd` and conflicts with `--harness`;
/// `--from`/`--to` are legal only with `--real`. Pure (no I/O, no exit) so the grammar is
/// unit-testable; `main` does the side effects.
fn parse_run_args(rest: &[&str]) -> Result<RunArgs, String> {
    let usage = || {
        "run [--harness <sma|macd|stage1-r>] [--macd] [--real <SYMBOL> [--from <ms>] [--to <ms>]] [--trace <name>]"
            .to_string()
    };
    let mut harness: Option<HarnessKind> = None;
    let mut macd_flag = false;
    let mut symbol: Option<String> = None;
    let mut from: Option<i64> = None;
    let mut to: Option<i64> = None;
    let mut trace: Option<String> = None;
    let mut tail = rest;
    while let Some((flag, t)) = tail.split_first() {
        match *flag {
            "--macd" if !macd_flag => {
                macd_flag = true;
                tail = t;
            }
            "--harness" if harness.is_none() => {
                let (value, t) = t.split_first().ok_or_else(usage)?;
                harness = Some(match *value {
                    "sma" => HarnessKind::Sma,
                    "macd" => HarnessKind::Macd,
                    "stage1-r" => HarnessKind::Stage1R,
                    _ => return Err(usage()),
                });
                tail = t;
            }
            "--real" if symbol.is_none() => {
                let (value, t) = t.split_first().ok_or_else(usage)?;
                if value.is_empty() {
                    return Err(usage());
                }
                symbol = Some((*value).to_string());
                tail = t;
            }
            "--from" if from.is_none() => {
                let (value, t) = t.split_first().ok_or_else(usage)?;
                from = Some(value.parse().map_err(|_| usage())?);
                tail = t;
            }
            "--to" if to.is_none() => {
                let (value, t) = t.split_first().ok_or_else(usage)?;
                to = Some(value.parse().map_err(|_| usage())?);
                tail = t;
            }
            "--trace" if trace.is_none() => {
                let (value, t) = t.split_first().ok_or_else(usage)?;
                trace = Some((*value).to_string());
                tail = t;
            }
            _ => return Err(usage()),
        }
    }
    // --macd is the alias; it conflicts with an explicit --harness.
    let harness = match (harness, macd_flag) {
        (Some(_), true) => return Err(usage()),
        (Some(h), false) => h,
        (None, true) => HarnessKind::Macd,
        (None, false) => HarnessKind::Sma,
    };
    // --from/--to require --real.
    if symbol.is_none() && (from.is_some() || to.is_some()) {
        return Err(usage());
    }
    let data = match symbol {
        Some(s) => RunData::Real { symbol: s, from, to },
        None => RunData::Synthetic,
    };
    Ok(RunArgs { harness, data, trace })
}

/// Route a parsed `run` invocation to its harness handler. The compile-time `match` IS
/// the selector — the harnesses are Rust-authored built-ins, picked by name (C9/C17).
fn run_dispatch(args: RunArgs) -> Result<RunReport, String> {
    let trace = args.trace.as_deref();
    Ok(match (args.harness, args.data) {
        (HarnessKind::Sma, RunData::Synthetic) => run_sample(trace),
        (HarnessKind::Macd, RunData::Synthetic) => run_macd(trace),
        (HarnessKind::Stage1R, data) => run_stage1_r(data, trace),
        (HarnessKind::Sma, RunData::Real { symbol, from, to }) => {
            run_sample_real(&symbol, from, to, trace)
        }
        (HarnessKind::Macd, RunData::Real { .. }) => {
            return Err("the macd harness has no --real form".to_string())
        }
    })
}

• Step 4: Replace the run argv arms in main()

Replace the five run arms (lines 1710-1722) with a single tail-parser arm:

        ["run", rest @ ..] => match parse_run_args(rest) {
            Ok(args) => match run_dispatch(args) {
                Ok(report) => println!("{}", report.to_json()),
                Err(msg) => {
                    eprintln!("aura: {msg}");
                    std::process::exit(2);
                }
            },
            Err(msg) => {
                eprintln!("aura: {msg}");
                std::process::exit(2);
            }
        },

(The ["chart", rest @ ..] arm and all others below stay unchanged.)

• Step 5: Update the USAGE string (line 1701-1702)

Change the leading run clause from aura run [--macd] [--trace <name>] | aura run --real <SYMBOL> [--from <ms>] [--to <ms>] [--trace <name>] to aura run [--harness <sma|macd|stage1-r>] [--macd] [--real <SYMBOL> [--from <ms>] [--to <ms>]] [--trace <name>].

• Step 6: Run the parser + selector tests, then the full cli_run suite

Run: cargo test -p aura-cli parse_run_args Expected: PASS — the three grammar tests.

Run: cargo test -p aura-cli run_harness Expected: PASS — run_harness_stage1_r_prints_an_r_block, run_harness_sma_is_pip_only_no_r_block, run_unknown_harness_exits_two.

Run: cargo test -p aura-cli --test cli_run Expected: PASS — every existing run/--macd/--real/--trace/chart/usage test stays green (the tokenizer accepts every form the old literal arms accepted, including the strict trailing-token and run --real paths), AND the new stage1_r_trace_persists_r_equity_and_charts_it round-trip passes, AND run_trace_index_carries_manifest_and_ordered_tap_list still pins the plain-run tap list to exactly ["equity","exposure"] (pip runs write no third tap).

---

Final gates (run after all tasks)

• Whole-workspace build

Run: cargo build --workspace Expected: clean.

• Whole-workspace test

Run: cargo test --workspace Expected: all green (the prior 500+ tests plus the new engine + cli tests).

• Lint

Run: cargo clippy --workspace --all-targets -- -D warnings Expected: clean (no unused imports left from the fixture rewires).