//! `aura` — the programmatic / CLI face of the engine (the surface the LLM and
//! automation drive: author a node, run a sim/sweep, emit structured metrics).
//!
//! The walking skeleton's closing seam: `aura run` bootstraps a built-in sample
//! signal-quality harness (synthetic source → SMA-cross → Exposure → SimBroker →
//! recording sinks), runs it deterministically (C1), and prints the run's
//! metrics + manifest (#6) as canonical JSON to stdout (the headline C14 move).
//!
//! `aura run --real <SYMBOL> [--from <ms>] [--to <ms>]` feeds that same harness
//! real M1 close bars, streamed lazily from the local data-server archive through
//! the #71 Source seam (`M1FieldSource`) instead of the synthetic stream — the
//! first real-data backtest from the CLI.

mod render;

use aura_core::{zip_params, Firing, Scalar, ScalarKind, Timestamp};
use aura_engine::{
    f64_field, monte_carlo, param_stability, summarize, walk_forward, window_of, Composite, Edge,
    FlatGraph, GraphBuilder, Harness, McAggregate, McFamily, RollMode, RunManifest, RunReport,
    SourceSpec, SweepFamily, SyntheticSpec, Target, VecSource, WalkForwardResult, WindowBounds,
    WindowRoller, WindowRun,
};
use aura_registry::{
    group_families, mc_member_reports, optimize, rank_by, sweep_member_reports,
    walkforward_member_reports, FamilyKind, Registry,
};
use aura_std::{Ema, Exposure, LinComb, Recorder, SimBroker, Sma, Sub};
use std::sync::mpsc::{self, Receiver};

/// The built-in synthetic price stream: rises through t=4 then reverses, so the
/// demo trace carries one exposure sign flip and a real drawdown (C22 populated
/// trace). Deterministic and fixed (C1).
fn synthetic_prices() -> Vec<(Timestamp, Scalar)> {
    [
        (1_i64, 1.0000_f64),
        (2, 1.0010),
        (3, 1.0030),
        (4, 1.0060),
        (5, 1.0040),
        (6, 1.0010),
        (7, 0.9990),
    ]
    .iter()
    .map(|&(t, p)| (Timestamp(t), Scalar::f64(p)))
    .collect()
}

/// A warm-up-adequate synthetic stream for the enriched sample/sweep: ~18 ticks
/// rising, falling, then rising again so the trend SMA spread and the MACD
/// EMA-of-EMA histogram both warm up and flip sign. It shares the proven warm-up
/// profile of `macd_prices` (the enriched sample embeds the same `macd` composite,
/// so it needs the same warm-up length); the flat `run_sample` keeps the shorter
/// `synthetic_prices`. Deterministic and fixed (C1).
fn showcase_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()
}

/// Bootstrap the sample signal-quality harness with two recording sinks (equity
/// tapped on the SimBroker, exposure tapped on the Exposure node). Rust-authored
/// wiring (C17/C20) over the raw bootstrap API — no builder DSL this cycle. The
/// price taps both SMAs and the broker's price slot (slot 1); exposure feeds the
/// broker's slot 0 (slot order is load-bearing — both are f64).
// The harness-plus-two-drained-sink-receivers tuple has exactly one call site
// (`run_sample`); a named type would be speculative abstraction this cycle.
#[allow(clippy::type_complexity)]
fn sample_harness() -> (
    Harness,
    Receiver<(Timestamp, Vec<Scalar>)>,
    Receiver<(Timestamp, Vec<Scalar>)>,
) {
    let (tx_eq, rx_eq) = mpsc::channel();
    let (tx_ex, rx_ex) = mpsc::channel();
    let f64_recorder_sig = || aura_engine::NodeSchema {
        inputs: vec![aura_engine::PortSpec { kind: ScalarKind::F64, firing: Firing::Any, name: "in".into() }],
        output: vec![],
        params: vec![],
    };
    let h = Harness::bootstrap(FlatGraph {
        nodes: vec![
            Box::new(Sma::new(2)),                                          // 0 fast SMA
            Box::new(Sma::new(4)),                                          // 1 slow SMA
            Box::new(Sub::new()),                                           // 2 spread
            Box::new(Exposure::new(0.5)),                                   // 3 exposure
            Box::new(SimBroker::new(0.0001)),                               // 4 sim-optimal broker
            Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx_eq)), // 5 equity sink
            Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx_ex)), // 6 exposure sink
        ],
        signatures: vec![
            Sma::builder().schema().clone(),
            Sma::builder().schema().clone(),
            Sub::builder().schema().clone(),
            Exposure::builder().schema().clone(),
            SimBroker::builder(0.0001).schema().clone(),
            f64_recorder_sig(),
            f64_recorder_sig(),
        ],
        sources: vec![SourceSpec {
            kind: ScalarKind::F64,
            targets: vec![
                Target { node: 0, slot: 0 },
                Target { node: 1, slot: 0 },
                Target { node: 4, slot: 1 }, // price into the broker's price slot
            ],
        }],
        edges: vec![
            Edge { from: 0, to: 2, slot: 0, from_field: 0 },
            Edge { from: 1, to: 2, slot: 1, from_field: 0 },
            Edge { from: 2, to: 3, slot: 0, from_field: 0 },
            Edge { from: 3, to: 4, slot: 0, from_field: 0 }, // exposure into broker slot 0
            Edge { from: 4, to: 5, slot: 0, from_field: 0 }, // equity -> sink 5
            Edge { from: 3, to: 6, slot: 0, from_field: 0 }, // exposure -> sink 6
        ],
    })
    .expect("valid sample signal-quality DAG");
    (h, rx_eq, rx_ex)
}

/// Build the sim-optimal `RunManifest`: the engine-external descriptor fields
/// (commit, seed-free synthetic run, broker label) are constant across the CLI's
/// built-in harnesses — only `params` and `window` vary. Centralizing the broker
/// label keeps it a single source (and a single edit point for #22's per-asset
/// pip work).
fn sim_optimal_manifest(
    params: Vec<(String, Scalar)>,
    window: (Timestamp, Timestamp),
    seed: u64,
) -> RunManifest {
    // The lossy f64 collapse lives here — the manifest field (the deferred typed
    // param-space precursor) owns its own lossiness; callers pass typed Scalars.
    let params = params.into_iter().map(|(n, s)| (n, scalar_as_param_f64(&s))).collect();
    RunManifest {
        commit: option_env!("AURA_COMMIT").unwrap_or("unknown").to_string(),
        params,
        window,
        seed,
        broker: "sim-optimal(pip_size=0.0001)".to_string(),
    }
}

/// Run the sample harness and fold it into a `RunReport` (drain both sinks →
/// `f64_field` → `summarize` → pair with a `RunManifest`). Pure and deterministic
/// (C1): the same build yields the same report.
fn run_sample() -> RunReport {
    let (mut h, rx_eq, rx_ex) = sample_harness();
    let sources: Vec<Box<dyn aura_engine::Source>> =
        vec![Box::new(VecSource::new(synthetic_prices()))];
    let window = window_of(&sources).expect("non-empty synthetic stream");
    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 equity = f64_field(&eq_rows, 0);
    let exposure = f64_field(&ex_rows, 0);
    let metrics = summarize(&equity, &exposure);

    RunReport {
        manifest: sim_optimal_manifest(
            vec![
                ("sma_fast".to_string(), Scalar::f64(2.0)),
                ("sma_slow".to_string(), Scalar::f64(4.0)),
                ("exposure_scale".to_string(), Scalar::f64(0.5)),
            ],
            window,
            0,
        ),
        metrics,
    }
}

/// `aura run --real <SYMBOL>`: run the built-in sample harness over real M1 close
/// bars streamed lazily from the local data-server archive through the #71 Source
/// seam (`M1FieldSource`, a `Box<dyn Source>`), not the synthetic `VecSource`. Same
/// fold as `run_sample`. The manifest window is read from a *separate* probe source
/// (a Source is single-pass), so the run source streams the window untouched.
/// A no-local-data condition (unknown symbol, or a window overlapping no file / no
/// bars) is a usage error: stderr + exit(2), not a panic.
fn run_sample_real(symbol: &str, from_ms: Option<i64>, to_ms: Option<i64>) -> RunReport {
    let (mut h, rx_eq, rx_ex) = sample_harness();
    let server = std::sync::Arc::new(data_server::DataServer::new(data_server::DEFAULT_DATA_PATH));
    let no_data = || -> ! {
        eprintln!(
            "aura: no local data for symbol '{symbol}' at {}",
            data_server::DEFAULT_DATA_PATH
        );
        std::process::exit(2)
    };
    if !server.has_symbol(symbol) {
        no_data();
    }
    let open = || aura_ingest::M1FieldSource::open(&server, symbol, from_ms, to_ms, aura_ingest::M1Field::Close);

    // Manifest window: drain a separate probe (single-pass Source) for first/last ts.
    let mut probe = match open() {
        Some(p) => p,
        None => no_data(),
    };
    let first = match aura_engine::Source::peek(&probe) {
        Some(t) => t,
        None => no_data(),
    };
    let mut last = first;
    while let Some((t, _)) = aura_engine::Source::next(&mut probe) {
        last = t;
    }
    let window = (first, last);

    let source: Box<dyn aura_engine::Source> = match open() {
        Some(s) => Box::new(s),
        None => no_data(),
    };
    h.run(vec![source]);

    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 equity = f64_field(&eq_rows, 0);
    let exposure = f64_field(&ex_rows, 0);
    let metrics = summarize(&equity, &exposure);

    RunReport {
        manifest: sim_optimal_manifest(
            vec![
                ("sma_fast".to_string(), Scalar::f64(2.0)),
                ("sma_slow".to_string(), Scalar::f64(4.0)),
                ("exposure_scale".to_string(), Scalar::f64(0.5)),
            ],
            window,
            0,
        ),
        metrics,
    }
}

/// Parse the `run --real` tail: a mandatory `<SYMBOL>` then zero-or-more
/// `--from <ms>` / `--to <ms>` pairs in any order. Pure (no I/O, no exit) so the
/// arg grammar is unit-testable; `main` does the side effects. An unknown token, a
/// flag without its value, a non-`i64` ms, or a repeated flag is an `Err` carrying
/// a short usage message.
fn parse_real_args(rest: &[&str]) -> Result<(String, Option<i64>, Option<i64>), String> {
    let usage = || "run --real <SYMBOL> [--from <ms>] [--to <ms>]".to_string();
    let (symbol, mut tail) = match rest.split_first() {
        Some((sym, t)) if !sym.is_empty() => (*sym, t),
        _ => return Err(usage()),
    };
    let mut from: Option<i64> = None;
    let mut to: Option<i64> = None;
    while let Some((flag, t)) = tail.split_first() {
        let (value, t) = t.split_first().ok_or_else(usage)?;
        let ms: i64 = value.parse().map_err(|_| usage())?;
        match *flag {
            "--from" if from.is_none() => from = Some(ms),
            "--to" if to.is_none() => to = Some(ms),
            _ => return Err(usage()),
        }
        tail = t;
    }
    Ok((symbol.to_string(), from, to))
}

/// The SMA-cross signal as a named composite (price -> fast/slow SMA -> spread).
/// CLI-local sample builder; the engine ships no sample (the duplication with
/// `blueprint.rs`'s test helper is the dedup tracked in #14). Value-empty: the SMA
/// lengths are injected at compile, not baked here.
fn sma_cross(name: &str) -> Composite {
    let mut g = GraphBuilder::new(name);
    let fast = g.add(Sma::builder().named("fast")); // fast SMA leg
    let slow = g.add(Sma::builder().named("slow")); // slow SMA leg
    let sub = g.add(Sub::builder());
    let price = g.input_role("price");
    g.feed(price, [fast.input("series"), slow.input("series")]);
    g.connect(fast.output("value"), sub.input("lhs"));
    g.connect(slow.output("value"), sub.input("rhs"));
    g.expose(sub.output("value"), "cross");
    g.build().expect("sample sma_cross wiring resolves")
}

/// The blended signal: a trend leg (SMA-cross) and a momentum leg (MACD), combined
/// by a weighted sum. A multiply-nested composite (root → signals → {trend,
/// momentum}); the blend is a multi-param node living inside it, with one weight
/// bound as a structural constant (so it drops out of the sweepable surface).
fn signals(name: &str) -> Composite {
    let mut g = GraphBuilder::new(name);
    let trend = g.add(sma_cross("trend")); // trend leg (one f64 "cross")
    let momentum = g.add(macd("momentum")); // momentum leg (3 outputs)
    // blend: Σ wᵢ·termᵢ over [trend.cross, momentum.histogram, momentum.signal].
    // weights[2] is bound (a fixed signal-line weight) → removed from param_space;
    // weights[0]/[1] stay tunable. `.named("blend")` makes the path signals.blend.*
    // (and renders the `blend:` prefix).
    let blend = g.add(
        LinComb::builder(3)
            .named("blend")
            .bind("weights[2]", Scalar::f64(0.5)),
    );
    let price = g.input_role("price");
    g.feed(price, [trend.input("price"), momentum.input("price")]);
    g.connect(trend.output("cross"), blend.input("term[0]")); // trend.cross        → blend.term[0]
    g.connect(momentum.output("histogram"), blend.input("term[1]")); // momentum.histogram → blend.term[1]
    g.connect(momentum.output("signal"), blend.input("term[2]")); // momentum.signal    → blend.term[2]
    g.expose(blend.output("value"), "signal");
    g.build().expect("sample signals wiring resolves")
}

/// The sample signal-quality blueprint (value-empty) **with its two recording
/// sinks reachable**: returns the equity + exposure receivers a per-point sweep
/// run drains (the eight free params — the trend SMA lengths, the momentum EMA
/// lengths, the two open blend weights, and the exposure scale — are injected at
/// compile via the point vector). The root harness of the sample topology, whose
/// signal is built by the nested `signals` composite; `build_sample` (the `aura
/// graph` entry) is expressed on top of it.
#[allow(clippy::type_complexity)]
fn sample_blueprint_with_sinks() -> (
    Composite,
    Receiver<(Timestamp, Vec<Scalar>)>,
    Receiver<(Timestamp, Vec<Scalar>)>,
) {
    let (tx_eq, rx_eq) = mpsc::channel();
    let (tx_ex, rx_ex) = mpsc::channel();
    let mut g = GraphBuilder::new("sample");
    let sig = g.add(signals("signals"));
    let exposure = g.add(Exposure::builder());
    let broker = g.add(SimBroker::builder(0.0001));
    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));
    let price = g.source_role("price", ScalarKind::F64);
    g.feed(price, [sig.input("price"), broker.input("price")]);
    g.connect(sig.output("signal"), exposure.input("signal")); // blended signal -> Exposure
    g.connect(exposure.output("exposure"), broker.input("exposure")); // exposure -> broker slot 0
    g.connect(broker.output("equity"), eq.input("col[0]")); // equity   -> sink
    g.connect(exposure.output("exposure"), ex.input("col[0]")); // exposure -> sink
    let bp = g.build().expect("sample blueprint wiring resolves");
    (bp, rx_eq, rx_ex)
}

/// The sample blueprint without its sink receivers — the `aura graph` render
/// entry, which never runs the graph (so the receivers are dropped).
fn build_sample() -> Composite {
    sample_blueprint_with_sinks().0
}

/// The built-in sample rendered by `aura graph`.
fn sample_blueprint() -> Composite {
    build_sample()
}

/// Coerce a sweep point's `Scalar` value to the manifest's `f64` param type.
/// A tuning grid carries only numeric params (`I64` lengths, `F64` scales).
fn scalar_as_param_f64(s: &Scalar) -> f64 {
    match s.kind() {
        ScalarKind::I64 => s.as_i64() as f64,
        ScalarKind::F64 => s.as_f64(),
        other => unreachable!("non-numeric sweep param: {other:?}"),
    }
}

/// Run the built-in sample over a small built-in grid (fast ∈ {2,3},
/// slow ∈ {4,5}, scale ∈ {0.5} — 4 points) and render one JSON line per point in
/// enumeration (odometer) order. Pure + deterministic (C1): the same build yields
/// the same report. Each point builds a fresh blueprint (fresh sink channels),
/// bootstraps it under the point vector, runs it, and folds the drained sinks to
/// metrics — the per-point closure the engine `sweep` drives disjointly.
fn sweep_family() -> SweepFamily {
    let bp = sample_blueprint_with_sinks().0;
    let space = bp.param_space();
    bp.axis("signals.trend.fast.length", [2, 3])
        .axis("signals.trend.slow.length", [4, 5])
        .axis("signals.momentum.fast.length", [2])
        .axis("signals.momentum.slow.length", [4])
        .axis("signals.momentum.signal.length", [3])
        .axis("signals.blend.weights[0]", [1.0])
        .axis("signals.blend.weights[1]", [1.0])
        .axis("exposure.scale", [0.5])
        .sweep(|point| {
            let (bp, rx_eq, rx_ex) = sample_blueprint_with_sinks();
            let mut h = bp
                .bootstrap_with_params(point.to_vec())
                .expect("grid points are kind-checked against param_space");
            let sources: Vec<Box<dyn aura_engine::Source>> =
                vec![Box::new(VecSource::new(showcase_prices()))];
            let window = window_of(&sources).expect("non-empty showcase stream");
            h.run(sources);
            let equity = f64_field(&rx_eq.try_iter().collect::<Vec<_>>(), 0);
            let exposure = f64_field(&rx_ex.try_iter().collect::<Vec<_>>(), 0);
            RunReport {
                manifest: sim_optimal_manifest(zip_params(&space, point), window, 0),
                metrics: summarize(&equity, &exposure),
            }
        })
        .expect("the built-in named grid matches the sample param-space")
}

/// Render a sweep family as one `RunReport` JSON line per point. Test helper:
/// production (`run_sweep`) renders *and* persists per point.
#[cfg(test)]
fn sweep_report() -> String {
    let mut out = String::new();
    for pt in &sweep_family().points {
        out.push_str(&pt.report.to_json());
        out.push('\n');
    }
    out
}

/// The default run registry: an append-only JSONL store under the current
/// working directory. (A project-configured runs-dir via `Aura.toml` is a later
/// refinement.)
fn default_registry() -> Registry {
    Registry::open("runs/runs.jsonl")
}

/// `aura sweep [--name <n>]`: run the built-in sweep, persist it as a *family*
/// (related records sharing one `family_id`, C18/C21) via `append_family`, and
/// print each point's record line carrying the assigned id.
fn run_sweep(name: &str) {
    let reg = default_registry();
    let family = sweep_family();
    let id = match reg.append_family(name, FamilyKind::Sweep, &sweep_member_reports(&family)) {
        Ok(id) => id,
        Err(e) => {
            eprintln!("aura: {e}");
            std::process::exit(2);
        }
    };
    for pt in &family.points {
        println!("{}", serde_json::json!({ "family_id": id, "report": pt.report }));
    }
}

/// `aura walkforward [--name <n>]`: run a built-in rolling walk-forward over the
/// sample blueprint + a synthetic windowed source. Per window: sweep the built-in
/// grid on the in-sample slice, optimize by total_pips (axis 2 inside axis 3,
/// where aura-cli bridges engine + registry), run the chosen params out-of-sample.
/// Persist the per-window OOS reports as a *family* (C18/C21) via `append_family`,
/// print each carrying the assigned id, then the stitched summary line.
/// Deterministic (C1).
fn run_walkforward(name: &str) {
    let reg = default_registry();
    let result = walkforward_family();
    let id =
        match reg.append_family(name, FamilyKind::WalkForward, &walkforward_member_reports(&result))
        {
            Ok(id) => id,
            Err(e) => {
                eprintln!("aura: {e}");
                std::process::exit(2);
            }
        };
    for w in &result.windows {
        println!("{}", serde_json::json!({ "family_id": id, "report": w.run.oos_report }));
    }
    println!("{}", walkforward_summary_json(&result));
}

/// The built-in rolling walk-forward: 24-bar in-sample, 12-bar out-of-sample,
/// stepping 12 (contiguous OOS tiling), over the 60-bar synthetic span -> 3
/// windows. Each window sweeps the built-in grid in-sample, optimizes by
/// total_pips (axis 2), and runs the chosen params out-of-sample.
fn walkforward_family() -> WalkForwardResult {
    let sources: Vec<Box<dyn aura_engine::Source>> =
        vec![Box::new(VecSource::new(walkforward_prices()))];
    let span = window_of(&sources).expect("non-empty synthetic stream");
    let roller = WindowRoller::new(span, 24, 12, 12, RollMode::Rolling)
        .expect("built-in walk-forward config fits the 60-bar synthetic span");
    walk_forward(roller, |w: WindowBounds| {
        let is_family = sweep_over(w.is.0, w.is.1);
        let best = optimize(&is_family, "total_pips").expect("total_pips is a known metric");
        let (oos_equity, oos_report) = run_oos(&best.params, w.oos.0, w.oos.1);
        WindowRun { chosen_params: best.params, oos_equity, oos_report }
    })
}

/// Sweep the built-in named grid over an in-sample window, sourcing the in-memory
/// windowed stream. Mirrors `sweep_family`, but windowed by `[from, to]`.
fn sweep_over(from: Timestamp, to: Timestamp) -> SweepFamily {
    let bp = sample_blueprint_with_sinks().0;
    let space = bp.param_space();
    bp.axis("signals.trend.fast.length", [2, 3])
        .axis("signals.trend.slow.length", [4, 5])
        .axis("signals.momentum.fast.length", [2])
        .axis("signals.momentum.slow.length", [4])
        .axis("signals.momentum.signal.length", [3])
        .axis("signals.blend.weights[0]", [1.0])
        .axis("signals.blend.weights[1]", [1.0])
        .axis("exposure.scale", [0.5])
        .sweep(|point| {
            let (bp, rx_eq, rx_ex) = sample_blueprint_with_sinks();
            let mut h = bp
                .bootstrap_with_params(point.to_vec())
                .expect("grid points are kind-checked against param_space");
            let sources: Vec<Box<dyn aura_engine::Source>> =
                vec![Box::new(walkforward_window_source(from, to))];
            let window = window_of(&sources).expect("non-empty in-sample window");
            h.run(sources);
            let equity = f64_field(&rx_eq.try_iter().collect::<Vec<_>>(), 0);
            let exposure = f64_field(&rx_ex.try_iter().collect::<Vec<_>>(), 0);
            RunReport {
                manifest: sim_optimal_manifest(zip_params(&space, point), window, 0),
                metrics: summarize(&equity, &exposure),
            }
        })
        .expect("the built-in named grid matches the sample param-space")
}

/// Run the chosen params over an out-of-sample window; return the recorded
/// pip-equity segment (for stitching) and the OOS RunReport (the C18 record).
fn run_oos(params: &[Scalar], from: Timestamp, to: Timestamp) -> (Vec<(Timestamp, f64)>, RunReport) {
    let (bp, rx_eq, rx_ex) = sample_blueprint_with_sinks();
    let space = bp.param_space();
    let mut h = bp
        .bootstrap_with_params(params.to_vec())
        .expect("chosen params are kind-checked against param_space");
    let sources: Vec<Box<dyn aura_engine::Source>> =
        vec![Box::new(walkforward_window_source(from, to))];
    let window = window_of(&sources).expect("non-empty out-of-sample window");
    h.run(sources);
    let equity = f64_field(&rx_eq.try_iter().collect::<Vec<_>>(), 0);
    let exposure = f64_field(&rx_ex.try_iter().collect::<Vec<_>>(), 0);
    let report = RunReport {
        manifest: sim_optimal_manifest(zip_params(&space, params), window, 0),
        metrics: summarize(&equity, &exposure),
    };
    (equity, report)
}

/// The walk-forward summary line: window count, stitched OOS total pips (the last
/// stitched-curve value), and the on-demand per-param stability. Canonical JSON
/// (C14).
fn walkforward_summary_json(result: &WalkForwardResult) -> String {
    let total = result.stitched_oos_equity.last().map(|&(_, v)| v).unwrap_or(0.0);
    serde_json::json!({
        "walkforward": {
            "windows": result.windows.len(),
            "stitched_total_pips": total,
            "param_stability": param_stability(result),
        }
    })
    .to_string()
}

/// A longer deterministic stream than `showcase_prices` — enough for several
/// IS/OOS windows with SMA warm-up. Seed-determined via `SyntheticSpec` (C1).
fn walkforward_prices() -> Vec<(Timestamp, Scalar)> {
    let spec = SyntheticSpec { start: 1.0, len: 60, step: 1 };
    let mut src = spec.source(7);
    let mut out = Vec::new();
    while let Some(item) = aura_engine::Source::next(&mut src) {
        out.push(item);
    }
    out
}

/// The in-memory windowed source the built-in demo uses (the firewall mapping to
/// `DataServer::stream_m1_windowed` is the real-data path; the demo stays in-memory,
/// mirroring `run_sweep`'s `showcase_prices`). Inclusive `[from, to]`.
fn walkforward_window_source(from: Timestamp, to: Timestamp) -> VecSource {
    VecSource::new(
        walkforward_prices()
            .into_iter()
            .filter(|&(t, _)| t >= from && t <= to)
            .collect(),
    )
}

/// Render the built-in walk-forward as the per-window OOS RunReport lines plus the
/// summary line — the `run_walkforward` shape minus registry persistence. Test
/// helper (mirrors `sweep_report`).
#[cfg(test)]
fn walkforward_report() -> String {
    let result = walkforward_family();
    let mut out = String::new();
    for w in &result.windows {
        out.push_str(&w.run.oos_report.to_json());
        out.push('\n');
    }
    out.push_str(&walkforward_summary_json(&result));
    out.push('\n');
    out
}

/// The built-in Monte-Carlo family: the sample harness over a fixed (empty) base
/// point, re-seeded across a built-in seed set — each seed a disjoint C1
/// realization of a synthetic price walk (C12 axis 4). Mirrors `sweep_family`,
/// varying the *seed* rather than a tuning param. The seed -> `Source`
/// construction lives inside the per-draw closure (eager-agnostic, #71).
fn mc_family() -> McFamily {
    let base_point: Vec<Scalar> = Vec::new();
    monte_carlo(&base_point, &[1, 2, 3], |seed, _base| {
        let (mut h, rx_eq, rx_ex) = sample_harness();
        let spec = SyntheticSpec { start: 1.0, len: 32, step: 1 };
        let sources: Vec<Box<dyn aura_engine::Source>> = vec![Box::new(spec.source(seed))];
        let window = window_of(&sources).expect("non-empty synthetic stream");
        h.run(sources);
        let equity = f64_field(&rx_eq.try_iter().collect::<Vec<_>>(), 0);
        let exposure = f64_field(&rx_ex.try_iter().collect::<Vec<_>>(), 0);
        RunReport {
            manifest: sim_optimal_manifest(
                vec![
                    ("sma_fast".to_string(), Scalar::f64(2.0)),
                    ("sma_slow".to_string(), Scalar::f64(4.0)),
                    ("exposure_scale".to_string(), Scalar::f64(0.5)),
                ],
                window,
                seed,
            ),
            metrics: summarize(&equity, &exposure),
        }
    })
}

/// Render an `McAggregate` as one canonical JSON line. `McAggregate` itself is not
/// `Serialize` (only its `MetricStats` fields are), so the line is built from the
/// three per-metric stat blocks.
fn mc_aggregate_json(agg: &McAggregate) -> String {
    serde_json::json!({
        "mc_aggregate": {
            "total_pips": agg.total_pips,
            "max_drawdown": agg.max_drawdown,
            "exposure_sign_flips": agg.exposure_sign_flips,
        }
    })
    .to_string()
}

/// `aura mc [--name <n>]`: run the built-in Monte-Carlo family, persist it to the
/// family store via `append_family` (C18/C21), print each draw's record line
/// (carrying the assigned `family_id`) plus the aggregate line.
fn run_mc(name: &str) {
    let reg = default_registry();
    let family = mc_family();
    let id = match reg.append_family(name, FamilyKind::MonteCarlo, &mc_member_reports(&family)) {
        Ok(id) => id,
        Err(e) => {
            eprintln!("aura: {e}");
            std::process::exit(2);
        }
    };
    for draw in &family.draws {
        println!(
            "{}",
            serde_json::json!({ "family_id": id, "seed": draw.seed, "report": draw.report })
        );
    }
    println!("{}", mc_aggregate_json(&family.aggregate));
}

/// Render the built-in Monte-Carlo family as the per-draw `RunReport` lines plus
/// the aggregate line — the `run_mc` shape minus registry persistence (no
/// `family_id`, which is store-assigned). Test helper, mirroring `sweep_report` /
/// `walkforward_report`: it carries the C1-determinism test of the family
/// computation, separate from the store-dependent id.
#[cfg(test)]
fn mc_report() -> String {
    let family = mc_family();
    let mut out = String::new();
    for draw in &family.draws {
        out.push_str(&draw.report.to_json());
        out.push('\n');
    }
    out.push_str(&mc_aggregate_json(&family.aggregate));
    out.push('\n');
    out
}

/// `aura runs list`: print every stored run record, in store (over-time) order.
fn runs_list() {
    for report in &load_runs_or_exit() {
        println!("{}", report.to_json());
    }
}

/// `aura runs rank <metric>`: print the stored runs best-first by `metric`.
fn runs_rank(metric: &str) {
    match rank_by(load_runs_or_exit(), metric) {
        Ok(ranked) => {
            for report in &ranked {
                println!("{}", report.to_json());
            }
        }
        Err(e) => {
            eprintln!("aura: {e}");
            std::process::exit(2);
        }
    }
}

/// `aura runs families`: one header line per stored family (id, kind, member
/// count), in first-seen store order.
fn runs_families() {
    let reg = default_registry();
    let members = match reg.load_family_members() {
        Ok(m) => m,
        Err(e) => {
            eprintln!("aura: {e}");
            std::process::exit(2);
        }
    };
    for fam in group_families(members) {
        println!(
            "{}",
            serde_json::json!({ "family_id": fam.id, "kind": fam.kind, "members": fam.members.len() })
        );
    }
}

/// `aura runs family <id> [rank <metric>]`: list one family's member reports in
/// ordinal order, or best-first by `metric`. An unknown id is an empty family
/// (prints nothing, exit 0, consistent with `runs list` over an empty store); an
/// unknown metric is a usage error (stderr + exit 2).
fn runs_family(id: &str, rank: Option<&str>) {
    let reg = default_registry();
    let members = match reg.load_family_members() {
        Ok(m) => m,
        Err(e) => {
            eprintln!("aura: {e}");
            std::process::exit(2);
        }
    };
    let Some(family) = group_families(members).into_iter().find(|f| f.id == id) else {
        return; // unknown family id: empty, exit 0
    };
    let reports: Vec<RunReport> = family.members.iter().map(|m| m.report.clone()).collect();
    let ordered = match rank {
        Some(metric) => match rank_by(reports, metric) {
            Ok(r) => r,
            Err(e) => {
                eprintln!("aura: {e}");
                std::process::exit(2);
            }
        },
        None => reports,
    };
    for report in &ordered {
        println!("{}", report.to_json());
    }
}

/// Load the registry or exit 2 with the error. A missing registry loads empty.
fn load_runs_or_exit() -> Vec<RunReport> {
    default_registry().load().unwrap_or_else(|e| {
        eprintln!("aura: {e}");
        std::process::exit(2);
    })
}

// --- MACD proof-of-concept (a richer, nested indicator + strategy) -----------

/// The MACD signal as a named composite: price → fast/slow `Ema` → the MACD line
/// (their spread) → a signal `Ema` of that line → the histogram (line − signal).
/// The composite exposes all **three MACD lines** as a named output record
/// (`macd`, `signal`, `histogram`); the strategy trades the histogram by reading
/// `from_field: 2`. A richer fixture than `sma_cross`: a nested EMA-of-EMA chain
/// with interior fan-out (the MACD line feeds *both* the signal EMA and the
/// histogram). Three `length` knobs (fast, slow, signal) are injected at compile
/// in node order; value-empty here.
fn macd(name: &str) -> Composite {
    let mut g = GraphBuilder::new(name);
    let fast = g.add(Ema::builder().named("fast")); // fast EMA
    let slow = g.add(Ema::builder().named("slow")); // slow EMA
    let line = g.add(Sub::builder()); // MACD line = fast − slow
    let signal = g.add(Ema::builder().named("signal")); // signal EMA of the MACD line
    let hist = g.add(Sub::builder()); // histogram = MACD line − signal
    let price = g.input_role("price");
    g.feed(price, [fast.input("series"), slow.input("series")]);
    g.connect(fast.output("value"), line.input("lhs")); // fast   → line
    g.connect(slow.output("value"), line.input("rhs")); // slow   → line
    g.connect(line.output("value"), signal.input("series")); // line → signal EMA
    g.connect(line.output("value"), hist.input("lhs")); // line → histogram
    g.connect(signal.output("value"), hist.input("rhs")); // signal → histogram
    g.expose(line.output("value"), "macd"); // the MACD line
    g.expose(signal.output("value"), "signal"); // the signal line
    g.expose(hist.output("value"), "histogram"); // the histogram
    g.build().expect("sample macd wiring resolves")
}

/// The MACD strategy blueprint (value-empty): the `macd` histogram → `Exposure` →
/// `SimBroker` → recording sinks. Channels are threaded so a run can drain the
/// sinks; `macd_blueprint` drops the receivers for the structural render.
fn macd_strategy_blueprint(
    tx_eq: mpsc::Sender<(Timestamp, Vec<Scalar>)>,
    tx_ex: mpsc::Sender<(Timestamp, Vec<Scalar>)>,
) -> Composite {
    let mut g = GraphBuilder::new("macd_strategy");
    let macd_node = g.add(macd("macd"));
    let exposure = g.add(Exposure::builder());
    let broker = g.add(SimBroker::builder(0.0001));
    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));
    let price = g.source_role("price", ScalarKind::F64);
    g.feed(price, [macd_node.input("price"), broker.input("price")]);
    g.connect(macd_node.output("histogram"), exposure.input("signal")); // histogram → Exposure
    g.connect(exposure.output("exposure"), broker.input("exposure")); // exposure  → broker slot 0
    g.connect(broker.output("equity"), eq.input("col[0]")); // equity    → sink
    g.connect(exposure.output("exposure"), ex.input("col[0]")); // exposure  → sink
    g.build().expect("macd_strategy wiring resolves")
}

/// The MACD strategy blueprint as a param-space fixture (receivers dropped, since
/// `param_space()` reads structure only, never runs the graph).
#[cfg(test)]
fn macd_blueprint() -> Composite {
    let (tx_eq, _rx_eq) = mpsc::channel();
    let (tx_ex, _rx_ex) = mpsc::channel();
    macd_strategy_blueprint(tx_eq, tx_ex)
}

/// The point vector for the MACD strategy, in `param_space()` slot order:
/// `[fast EMA length, slow EMA length, signal EMA length, exposure scale]`. Short
/// windows so the 7-tick synthetic stream still produces a non-trivial trace
/// (conventional MACD is 12/26/9, meaningless on 7 points).
fn macd_point() -> Vec<Scalar> {
    vec![Scalar::i64(2), Scalar::i64(4), Scalar::i64(3), Scalar::f64(0.5)]
}

/// A longer synthetic stream than the SMA sample's 7 ticks: MACD's EMAs each warm
/// up over their `length`, so the stream rises, falls, then rises again to give the
/// histogram room to flip sign more than once *after* warm-up. Deterministic (C1).
fn macd_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()
}

/// Run the MACD strategy: compile the nested composite blueprint to a flat harness
/// (the same bootstrap path the SMA sample's compiled view uses), drive it on the
/// synthetic stream, and fold both sinks into a `RunReport`. Pure and
/// deterministic (C1).
fn run_macd() -> RunReport {
    let (tx_eq, rx_eq) = mpsc::channel();
    let (tx_ex, rx_ex) = mpsc::channel();
    let flat = macd_strategy_blueprint(tx_eq, tx_ex)
        .compile_with_params(&macd_point())
        .expect("valid macd blueprint");
    let mut h = Harness::bootstrap(flat).expect("valid macd harness");

    let sources: Vec<Box<dyn aura_engine::Source>> =
        vec![Box::new(VecSource::new(macd_prices()))];
    let window = window_of(&sources).expect("non-empty macd stream");
    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 equity = f64_field(&eq_rows, 0);
    let exposure = f64_field(&ex_rows, 0);
    let metrics = summarize(&equity, &exposure);

    RunReport {
        manifest: sim_optimal_manifest(
            vec![
                ("ema_fast".to_string(), Scalar::f64(2.0)),
                ("ema_slow".to_string(), Scalar::f64(4.0)),
                ("ema_signal".to_string(), Scalar::f64(3.0)),
                ("exposure_scale".to_string(), Scalar::f64(0.5)),
            ],
            window,
            0,
        ),
        metrics,
    }
}

const USAGE: &str =
    "usage: aura run [--macd] | aura run --real <SYMBOL> [--from <ms>] [--to <ms>] | aura graph | aura sweep [--name <n>] | aura mc [--name <n>] | aura walkforward [--name <n>] | aura runs list | aura runs rank <metric> | aura runs families | aura runs family <id> [rank <metric>]";

fn main() {
    // Collect argv and match the whole vector: every accepted form is exhaustive,
    // so an unexpected trailing token falls through to the usage-error path rather
    // than masquerading as a successful run (#16 strict reading).
    let args: Vec<String> = std::env::args().skip(1).collect();
    match args.iter().map(String::as_str).collect::<Vec<_>>().as_slice() {
        ["run"] => println!("{}", run_sample().to_json()),
        ["run", "--macd"] => println!("{}", run_macd().to_json()),
        ["run", "--real", rest @ ..] => match parse_real_args(rest) {
            Ok((sym, from, to)) => println!("{}", run_sample_real(&sym, from, to).to_json()),
            Err(msg) => {
                eprintln!("aura: {msg}");
                std::process::exit(2);
            }
        },
        ["graph"] => print!("{}", render::render_html(&sample_blueprint())),
        ["sweep"] => run_sweep("sweep"),
        ["sweep", "--name", n] => run_sweep(n),
        ["walkforward"] => run_walkforward("walkforward"),
        ["walkforward", "--name", n] => run_walkforward(n),
        ["mc"] => run_mc("mc"),
        ["mc", "--name", n] => run_mc(n),
        ["runs", "list"] => runs_list(),
        ["runs", "rank", metric] => runs_rank(metric),
        ["runs", "families"] => runs_families(),
        ["runs", "family", id] => runs_family(id, None),
        ["runs", "family", id, "rank", metric] => runs_family(id, Some(metric)),
        ["--help"] | ["-h"] => println!("{USAGE}"),
        _ => {
            eprintln!("aura: {USAGE}");
            std::process::exit(2);
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn walkforward_report_is_deterministic() {
        // spec §Testing 10: the built-in WFO render is byte-identical across two
        // calls (C1).
        assert_eq!(walkforward_report(), walkforward_report());
    }

    #[test]
    fn walkforward_report_has_one_oos_line_per_window_plus_summary() {
        // spec §Testing 11: N per-window OOS RunReport lines + one summary line.
        let out = walkforward_report();
        let lines: Vec<&str> = out.lines().collect();
        assert_eq!(lines.len(), 4); // built-in roll = 3 windows + 1 summary
        assert!(lines[3].contains(r#""walkforward""#), "summary line: {}", lines[3]);
        for line in &lines[..3] {
            assert!(
                line.contains(r#""manifest""#) && line.contains(r#""metrics""#),
                "expected an OOS RunReport line, got: {line}",
            );
        }
    }

    /// The drained sink trace of a seeded run — the recorded rows of the equity
    /// and exposure sinks. Compared row-for-row so the C1 seed-determinism
    /// property is tested at the trace level (strictly stronger than the folded
    /// 3-field metrics). `PartialEq` not `Eq`: `Scalar` carries `f64`.
    #[derive(Debug, PartialEq)]
    struct SeededTrace {
        equity: Vec<(Timestamp, Vec<Scalar>)>,
        exposure: Vec<(Timestamp, Vec<Scalar>)>,
    }

    /// A seeded run of the sample harness: the synthetic stream is generated
    /// from `seed`, that same seed is recorded into the manifest, and the
    /// drained sink trace is returned alongside the report. Every byte of both
    /// is a function of `seed`.
    fn run_sample_seeded(seed: u64) -> (RunReport, SeededTrace) {
        let (mut h, rx_eq, rx_ex) = sample_harness();
        let spec = SyntheticSpec { start: 1.0, len: 64, step: 1 };
        let window = (Timestamp(1), Timestamp((spec.len as i64 - 1) * spec.step + 1));
        h.run(vec![Box::new(spec.source(seed))]);
        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 metrics = summarize(&f64_field(&eq_rows, 0), &f64_field(&ex_rows, 0));
        let report = RunReport {
            manifest: sim_optimal_manifest(
                vec![
                    ("sma_fast".to_string(), Scalar::f64(2.0)),
                    ("sma_slow".to_string(), Scalar::f64(4.0)),
                    ("exposure_scale".to_string(), Scalar::f64(0.5)),
                ],
                window,
                seed,
            ),
            metrics,
        };
        (report, SeededTrace { equity: eq_rows, exposure: ex_rows })
    }

    #[test]
    fn same_seed_bit_identical_trace() {
        // Bit-identical sink trace for a fixed seed (acceptance bullet 1, C1).
        let (_, trace_a) = run_sample_seeded(42);
        let (_, trace_b) = run_sample_seeded(42);
        assert_eq!(trace_a, trace_b);
    }

    #[test]
    fn different_seed_different_trace() {
        // Different seeds perturb the trace (acceptance bullet 2).
        let (a, _) = run_sample_seeded(1);
        let (b, _) = run_sample_seeded(2);
        assert_ne!(a.metrics, b.metrics);
    }

    #[test]
    fn seed_recorded_in_manifest() {
        // The seed that drove the run is recorded (acceptance bullet 3).
        let (report, _) = run_sample_seeded(7);
        assert_eq!(report.manifest.seed, 7);
    }

    #[test]
    fn sample_blueprint_with_sinks_bootstraps_runs_and_drains() {
        // the factory returns the two Recorder receivers (build_sample drops them),
        // so a caller can bootstrap one point, run it, and drain both sinks.
        let (bp, rx_eq, rx_ex) = sample_blueprint_with_sinks();
        let mut h = bp
            .with("signals.trend.fast.length", 2)
            .with("signals.trend.slow.length", 4)
            .with("signals.momentum.fast.length", 2)
            .with("signals.momentum.slow.length", 4)
            .with("signals.momentum.signal.length", 3)
            .with("signals.blend.weights[0]", 1.0)
            .with("signals.blend.weights[1]", 1.0)
            .with("exposure.scale", 0.5)
            .bootstrap()
            .expect("sample blueprint compiles under a valid point");
        h.run(vec![Box::new(VecSource::new(showcase_prices()))]);
        assert!(!rx_eq.try_iter().collect::<Vec<_>>().is_empty(), "equity sink drained empty");
        assert!(!rx_ex.try_iter().collect::<Vec<_>>().is_empty(), "exposure sink drained empty");
    }

    #[test]
    fn sweep_report_renders_four_points_in_odometer_order() {
        let out = sweep_report();
        let lines: Vec<&str> = out.lines().collect();
        assert_eq!(lines.len(), 4, "one JSON line per grid point; got: {out:?}");
        // each line is a full RunReport; the commit is the real git HEAD
        // (volatile), so pin the per-point manifest params (odometer order, last
        // axis fastest) + the metric keys, not the commit value.
        for line in &lines {
            assert!(line.starts_with(r#"{"manifest":{"commit":""#), "not a RunReport: {line}");
        }
        assert!(lines[0].contains(r#""params":[["signals.trend.fast.length",2.0],["signals.trend.slow.length",4.0],["signals.momentum.fast.length",2.0],["signals.momentum.slow.length",4.0],["signals.momentum.signal.length",3.0],["signals.blend.weights[0]",1.0],["signals.blend.weights[1]",1.0],["exposure.scale",0.5]]"#), "line0: {}", lines[0]);
        assert!(lines[1].contains(r#""params":[["signals.trend.fast.length",2.0],["signals.trend.slow.length",5.0],["signals.momentum.fast.length",2.0],["signals.momentum.slow.length",4.0],["signals.momentum.signal.length",3.0],["signals.blend.weights[0]",1.0],["signals.blend.weights[1]",1.0],["exposure.scale",0.5]]"#), "line1: {}", lines[1]);
        assert!(lines[2].contains(r#""params":[["signals.trend.fast.length",3.0],["signals.trend.slow.length",4.0],["signals.momentum.fast.length",2.0],["signals.momentum.slow.length",4.0],["signals.momentum.signal.length",3.0],["signals.blend.weights[0]",1.0],["signals.blend.weights[1]",1.0],["exposure.scale",0.5]]"#), "line2: {}", lines[2]);
        assert!(lines[3].contains(r#""params":[["signals.trend.fast.length",3.0],["signals.trend.slow.length",5.0],["signals.momentum.fast.length",2.0],["signals.momentum.slow.length",4.0],["signals.momentum.signal.length",3.0],["signals.blend.weights[0]",1.0],["signals.blend.weights[1]",1.0],["exposure.scale",0.5]]"#), "line3: {}", lines[3]);
        for line in &lines {
            assert!(line.contains(r#""total_pips":"#), "missing total_pips: {line}");
            assert!(line.contains(r#""max_drawdown":"#), "missing max_drawdown: {line}");
            assert!(line.contains(r#""exposure_sign_flips":"#), "missing flips: {line}");
            assert!(line.ends_with('}'), "line not closed: {line}");
        }
    }

    #[test]
    fn sweep_report_is_deterministic() {
        // C1 at the CLI edge: the same build yields a bit-identical report.
        assert_eq!(sweep_report(), sweep_report());
    }

    #[test]
    fn mc_report_is_deterministic_and_one_line_per_seed() {
        // C1 at the CLI edge: the family computation renders bit-identically.
        assert_eq!(mc_report(), mc_report());
        let out = mc_report();
        let lines: Vec<&str> = out.lines().collect();
        // three seeds -> three member lines + one aggregate line
        assert_eq!(lines.len(), 4, "expected 3 members + 1 aggregate: {out}");
        for line in &lines[..3] {
            assert!(line.contains(r#""total_pips":"#), "member line missing metrics: {line}");
        }
        assert!(lines[3].contains(r#""mc_aggregate":"#), "missing aggregate line: {}", lines[3]);
    }

    #[test]
    fn cli_families_persist_and_round_trip_per_kind() {
        use aura_registry::{
            group_families, mc_member_reports, sweep_member_reports, walkforward_member_reports,
            FamilyKind, Registry,
        };
        let dir = std::env::temp_dir().join(format!("aura-cli-fam-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&dir);
        std::fs::create_dir_all(&dir).expect("temp dir");
        let reg = Registry::open(dir.join("runs.jsonl"));

        // the exact persist chain `run_sweep`/`run_mc`/`run_walkforward` use, against
        // a fresh temp store (the run_* fns themselves bind `default_registry()`):
        // engine family -> per-kind extractor -> append_family.
        let sid = reg
            .append_family("sweep", FamilyKind::Sweep, &sweep_member_reports(&sweep_family()))
            .expect("sweep family");
        let mid = reg
            .append_family("mc", FamilyKind::MonteCarlo, &mc_member_reports(&mc_family()))
            .expect("mc family");
        let wid = reg
            .append_family(
                "walkforward",
                FamilyKind::WalkForward,
                &walkforward_member_reports(&walkforward_family()),
            )
            .expect("walkforward family");
        assert_eq!((sid.as_str(), mid.as_str(), wid.as_str()), ("sweep-0", "mc-0", "walkforward-0"));

        let families = group_families(reg.load_family_members().expect("load"));
        assert_eq!(families.len(), 3);
        let by_id = |id: &str| families.iter().find(|f| f.id == id).expect("family present");
        assert_eq!(by_id("sweep-0").kind, FamilyKind::Sweep);
        assert_eq!(by_id("mc-0").kind, FamilyKind::MonteCarlo);
        assert_eq!(by_id("mc-0").members.len(), 3); // 3 seeds
        assert_eq!(by_id("walkforward-0").kind, FamilyKind::WalkForward);
        assert_eq!(by_id("walkforward-0").members.len(), 3); // 3 windows
        let _ = std::fs::remove_dir_all(&dir);
    }

    #[test]
    fn run_macd_compiles_from_nested_composite_and_is_deterministic() {
        // the MACD strategy authors a nested EMA-of-EMA composite, compiles it to a
        // flat runnable harness (the call not panicking proves the compile+bootstrap
        // path), and runs it. C1 determinism: two runs are bit-identical.
        let r1 = run_macd();
        let r2 = run_macd();
        assert_eq!(r1.metrics, r2.metrics);
        assert_eq!(r1.to_json(), r2.to_json());

        // the synthetic stream is carried end-to-end and the trace is well-formed.
        let (from, to) = r1.manifest.window;
        assert_eq!((from.0, to.0), (1, 18));
        assert!(r1.metrics.total_pips.is_finite(), "macd pips must be finite: {:?}", r1.metrics);
        assert!(r1.metrics.max_drawdown >= 0.0, "drawdown is non-negative: {:?}", r1.metrics);
        // after warm-up the EMA-of-EMA histogram crosses zero, so the strategy
        // reverses exposure at least once — a genuinely non-trivial trace.
        assert!(
            r1.metrics.exposure_sign_flips >= 1,
            "macd trace should flip exposure: {:?}",
            r1.metrics
        );
    }

    /// E2E acceptance (#41 / spec 0019, the worked example): the real MACD strategy
    /// blueprint's swept param surface qualifies the three otherwise-indistinguishable
    /// EMA `length` slots by node name to `macd.fast.length` / `macd.slow.length` /
    /// `macd.signal.length` — the named composite boundary visible end-to-end through
    /// `param_space()`, with the slot count and order unchanged (C23 — node names are
    /// non-load-bearing: every interior slot stays sweepable, the `exposure.scale`
    /// knob is unaffected).
    #[test]
    fn macd_param_space_surfaces_the_three_named_legs() {
        let names: Vec<String> =
            macd_blueprint().param_space().into_iter().map(|p| p.name).collect();
        // three named composite-interior slots, in declared (fast, slow, signal)
        // order, then the strategy-level Exposure `scale` (a root-level leaf).
        assert_eq!(
            names,
            vec![
                "macd.fast.length".to_string(),
                "macd.slow.length".to_string(),
                "macd.signal.length".to_string(),
                "exposure.scale".to_string(),
            ],
            "MACD param surface must expose the three named EMA lengths + scale",
        );
    }

    /// `aura run --real <SYMBOL>` dogfoods the #71 streaming Source seam: the same
    /// built-in sample signal-quality harness, but fed real M1 **close** bars
    /// streamed lazily through `aura_ingest::M1FieldSource` (a `Box<dyn Source>`),
    /// not synthetic `VecSource` prices. The property: over a bounded real window,
    /// `run_sample_real` yields a `RunReport` whose `total_pips` is finite and is
    /// C1-deterministic — two runs of the same window are bit-identical JSON.
    ///
    /// Gated like the ingest `streaming_seam` test: skip (early return) when the
    /// local Pepperstone archive is absent, so the spec never fails on a machine
    /// without the data. Uses the verified bounded 2006-08 `AAPL.US` window (the
    /// same FROM_MS/TO_MS the ingest seam test drives) so the two-run determinism
    /// check stays fast.
    #[test]
    fn run_sample_real_streams_real_close_bars_deterministically() {
        // Same bounded 2006-08 inclusive-ms window the ingest streaming_seam test
        // uses; AAPL.US M1 data is present there on a data-bearing machine.
        const SYMBOL: &str = "AAPL.US";
        const FROM_MS: i64 = 1_154_390_400_000;
        const TO_MS: i64 = 1_157_068_799_999;

        // Mirror skip_if_no_data: never fail where the local archive is absent.
        let server = std::sync::Arc::new(data_server::DataServer::new(
            data_server::DEFAULT_DATA_PATH,
        ));
        if !server.has_symbol(SYMBOL) {
            eprintln!(
                "skip: no local data at {} (symbol {SYMBOL} absent)",
                data_server::DEFAULT_DATA_PATH
            );
            return;
        }

        // The headline: a real-data run over the bounded window yields a finite,
        // C1-deterministic RunReport.
        let r1 = run_sample_real(SYMBOL, Some(FROM_MS), Some(TO_MS));
        let r2 = run_sample_real(SYMBOL, Some(FROM_MS), Some(TO_MS));

        assert!(
            r1.metrics.total_pips.is_finite(),
            "real-data run must yield finite pips: {:?}",
            r1.metrics
        );
        // C1 at the CLI edge: the same real window streamed twice is bit-identical.
        assert_eq!(
            r1.to_json(),
            r2.to_json(),
            "two real-data runs of the same window must be bit-identical (C1)"
        );
    }

    /// `parse_real_args` accepts a bare symbol (no window) and a full
    /// symbol + `--from`/`--to` pair in any order, and rejects a flag missing its
    /// value — the pure arg grammar `main` relies on for `run --real`.
    #[test]
    fn parse_real_args_accepts_symbol_and_optional_window() {
        assert_eq!(parse_real_args(&["EURUSD"]), Ok(("EURUSD".to_string(), None, None)));
        assert_eq!(
            parse_real_args(&["EURUSD", "--from", "100", "--to", "200"]),
            Ok(("EURUSD".to_string(), Some(100), Some(200)))
        );
        // flags in any order
        assert_eq!(
            parse_real_args(&["EURUSD", "--to", "200", "--from", "100"]),
            Ok(("EURUSD".to_string(), Some(100), Some(200)))
        );
        // a flag without its value, an empty symbol, and a non-numeric ms all reject
        assert!(parse_real_args(&["EURUSD", "--from"]).is_err());
        assert!(parse_real_args(&[]).is_err());
        assert!(parse_real_args(&["EURUSD", "--from", "notanumber"]).is_err());
    }

    #[test]
    fn run_sample_is_deterministic_and_non_trivial() {
        let r1 = run_sample();
        let r2 = run_sample();
        // C1 determinism: two runs are bit-identical (metrics + rendered JSON).
        assert_eq!(r1.metrics, r2.metrics);
        assert_eq!(r1.to_json(), r2.to_json());

        let m = &r1.metrics;
        // exactly one exposure sign flip in the demo trace (rises then reverses).
        assert_eq!(m.exposure_sign_flips, 1);
        // a non-trivial, populated trace: a real drawdown.
        assert!(m.max_drawdown > 0.0);
        // hand-computed magnitudes for the chosen stream (float tolerance; the
        // computation's dust is ~1e-15).
        assert!(
            (m.max_drawdown - 0.17).abs() < 1e-9,
            "max_drawdown = {}",
            m.max_drawdown
        );
        assert!(
            (m.total_pips - (-0.13)).abs() < 1e-9,
            "total_pips = {}",
            m.total_pips
        );

        // manifest carries the sample's known configuration.
        let (from, to) = r1.manifest.window;
        assert_eq!((from.0, to.0), (1, 7));
        // commit is the build's git identity (or the no-git "unknown" fallback);
        // either way it is non-empty and fixed at compile time, so it is stable
        // across runs of the same build (C1 determinism, already asserted above
        // via `to_json()`).
        assert!(!r1.manifest.commit.is_empty());
        assert_eq!(r1.manifest.commit, r2.manifest.commit);
    }
}