//! `aura` — the programmatic / CLI face of the engine. Topology lives only as
//! data (#159): `aura run`/`sweep`/`walkforward`/`mc` load a serialized blueprint
//! (or the `--strategy r-sma` sugar) and print canonical JSON metrics/manifests;
//! this binary authors no built-in harness.

mod binding;
mod render;
mod graph_construct;
mod project;
mod campaign_run;
mod research_docs;
mod scaffold;
mod verb_sugar;
use render::{ChartData, ChartMeta, ChartMode, ReduceKind, Series};

use aura_core::{zip_params, Cell, Firing, ParamSpec, PrimitiveBuilder, Scalar, ScalarKind, Timestamp};
use aura_composites::{cost_graph, risk_executor, StopRule};
use aura_engine::{
    blueprint_from_json, blueprint_to_json, f64_field, join_on_ts, monte_carlo, param_stability,
    r_metrics_from_rs, summarize, summarize_r, walk_forward, window_of, BindError, BlueprintNode,
    Composite, FamilySelection, GraphBuilder, Harness, JoinedRow, McAggregate,
    McFamily, RBootstrap, RollMode, RunManifest, RunMetrics, RunReport, SelectionMode,
    SweepFamily, SweepPoint, SyntheticSpec, VecSource, WalkForwardResult,
    WindowBounds, WindowRoller, WindowRun,
};
use aura_registry::{
    check_r_metric, group_families, mc_member_reports, optimize_deflated,
    optimize_plateau, rank_by, sweep_member_reports, walkforward_member_reports, Family, FamilyKind,
    FamilyMember, Generalization, NameKind, PlateauMode, Registry, RunTraces,
    DEFLATION_BLOCK_LEN, DEFLATION_N_RESAMPLES,
};
use aura_std::{
    cost_port, GatedRecorder, LinComb, Recorder, RollingMax, RollingMin,
    SeriesReducer, SimBroker, Sub, GEOMETRY_WIDTH, PM_FIELD_NAMES,
    PM_RECORD_KINDS,
};
// `std_vocabulary` is now only reached through `project::Env::resolve` in production
// code; the test module still builds reference blueprints against it directly, so
// the import is test-only.
#[cfg(test)]
use aura_std::std_vocabulary;
// `Delay` is now only used by the test-only `r_breakout_signal` carve (#159 cut 2's
// fused-builder retirement dropped the production caller); the import is test-only.
#[cfg(test)]
use aura_std::Delay;
// `Scale` is only used by the test-only `r_meanrev_signal` carve (#159 cut 3; the
// band half-width uses `Scale` in place of `LinComb(1)`) — no production caller, so
// the import is test-only, mirroring `Delay` above.
#[cfg(test)]
use aura_std::Scale;
// `Add`/`Gt`/`Latch`/`Mul`/`Sqrt` are now only reached from the test-only
// `r_breakout_signal`/`r_meanrev_signal` carves — #159 cut 3's retirement of the
// last fused mean-reversion builder dropped the last production caller, mirroring
// `Delay`/`Scale` above.
#[cfg(test)]
use aura_std::{Add, Gt, Latch, Mul, Sqrt};
// `ColumnarTrace` is now also reached from `run_signal_r`'s tap-persistence
// path, so it is a production import (no longer test-only).
use aura_engine::ColumnarTrace;
// `Bias`/`Ema`/`Sma` are only reached from the test module; the imports are
// test-only, mirroring `Delay`/`Scale` above.
#[cfg(test)]
use aura_std::{Bias, Ema, Sma};
#[cfg(test)]
use aura_std::{ConstantCost, VolSlippageCost};
use std::sync::mpsc;
use std::sync::LazyLock;
use std::collections::{BTreeMap, BTreeSet, HashSet};
use clap::{Args, Parser, Subcommand};

/// The pip size the built-in *synthetic* families (sweep/walkforward/mc over a
/// blueprint) run at: a 5-decimal FX major (`EURUSD`-shaped). The synthetic streams
/// carry no instrument, so there is no recorded geometry sidecar to thread; a
/// single named source keeps the broker's divisor (`SimBroker::builder`) and the
/// recorded broker label (`sim_optimal_manifest`) in lockstep, so they cannot
/// silently drift apart. The real path threads the sidecar's looked-up `pip_size`
/// instead.
const SYNTHETIC_PIP_SIZE: f64 = 0.0001;

/// Real walk-forward roller sizes (Fork D/F). `WindowRoller` takes sizes in the
/// stream's epoch-unit; for real M1 that is nanoseconds. A classic 3-month
/// in-sample / 1-month out-of-sample / 1-month step (contiguous OOS tiling).
const WF_DAY_NS: i64 = 86_400_000_000_000;
const WF_REAL_IS_NS: i64 = 90 * WF_DAY_NS;
const WF_REAL_OOS_NS: i64 = 30 * WF_DAY_NS;
const WF_REAL_STEP_NS: i64 = 30 * WF_DAY_NS;

/// The winner-selection objective for walk-forward's per-window IS refit — the
/// deflation-aware SQN variant (#144 default). Named once so the three call sites
/// (the inline r-sma roller, the inline blueprint roller, and the dissolved
/// campaign sugar's bridge) cannot drift apart on the token.
const WINNER_SELECTION_METRIC: &str = "sqn_normalized";

/// Fixed RNG seed for the trials-deflation reality-check bootstrap in walk-forward
/// winner selection. Recorded on each winner's manifest (so `overfit_probability`
/// is reproducible by re-run); a CLI flag for it is a deferred refinement. The
/// resample count and block length are the shared `aura_registry::DEFLATION_*`.
const DEFLATION_SEED: u64 = 0xDEF1_A7ED;

/// A warm-up-adequate synthetic stream (~18 ticks rising, falling, then rising
/// again) used as `DataSource::Synthetic`'s full-window stream for the built-in
/// blueprint/campaign sweep, walk-forward, and MC families. 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()
}

/// 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 per-asset pip
/// work): the `pip_size` it renders is the one its caller ran the broker at.
fn sim_optimal_manifest(
    params: Vec<(String, Scalar)>,
    window: (Timestamp, Timestamp),
    seed: u64,
    pip_size: f64,
) -> RunManifest {
    // Typed params pass straight through: the manifest carries self-describing
    // Scalars, so a length stays `i64` and a scale `f64` in the record.
    RunManifest {
        commit: ENGINE_COMMIT.to_string(),
        params,
        defaults: Vec::new(),
        window,
        seed,
        broker: format!("sim-optimal(pip_size={pip_size})"),
        selection: None,
        instrument: None,
        topology_hash: None,
        project: None,
    }
}

/// The wrap-prefixed BOUND-param defaults of `signal` (#249): every
/// `bound_param_space()` entry as a `(<signal.name()>.<param>, value)` pair, in
/// `bound_param_space()` order — the `RunManifest.defaults` field. Must be read
/// off `signal` BEFORE it is consumed by `wrap_r`/reopening: an axis-reopened
/// bound param has already left `bound_param_space()` by construction
/// (`Composite::reopen` forgets the bound value), so a caller that reopens
/// overrides before calling this naturally excludes them — no separate filter
/// needed. Same prefixing rule as `wrapped_bound_names`, kept separate because
/// that helper discards the value this one needs.
fn wrapped_bound_defaults(signal: &Composite) -> Vec<(String, Scalar)> {
    let prefix = format!("{}.", signal.name());
    signal
        .bound_param_space()
        .into_iter()
        .map(|b| (format!("{prefix}{}", b.name), b.value))
        .collect()
}

/// Map any byte outside the portable directory-name charset `[A-Za-z0-9._-]` to
/// `_`. The single source of filesystem-portability for an on-disk path component
/// (valid on Linux / Windows / macOS, also URL-path- and cloud-sync-safe). Used by
/// `campaign_run`'s per-window member naming.
fn sanitize_component(s: &str) -> String {
    s.chars()
        .map(|c| if c.is_ascii_alphanumeric() || matches!(c, '.' | '_' | '-') { c } else { '_' })
        .collect()
}

/// Render a scalar value case-lessly: integers/timestamps as decimal digits, bool
/// as `true`/`false`, f64 via Rust's `Display` (decimal, shortest round-trip, NO
/// scientific notation). Case-less rendering is what keeps two members of one
/// family from ever differing only by letter case (case-insensitive-FS safety).
fn render_value(v: &Scalar) -> String {
    match v {
        Scalar::I64(n) => n.to_string(),
        Scalar::F64(x) => x.to_string(),
        Scalar::Bool(b) => b.to_string(),
        Scalar::Timestamp(t) => t.0.to_string(),
    }
}

/// Default decimation budget: target horizontal buckets. ~2000 buckets ⇒ ≤ ~4000
/// spine slots (min+max per bucket) — a few-thousand-point page regardless of the
/// underlying multi-year M1 point count.
const CHART_DECIMATE_BUCKETS: usize = 2000;

/// Per-tap decimation kind (#111): the bounded exposure stream (C10, f64 ∈ [-1,+1])
/// reduces by per-bucket mean, so its net/duty-cycle level survives decimation
/// instead of collapsing to a -1..+1 band (every bucket of a multi-year exposure
/// straddles many sign flips, so min/max would be ±1 everywhere). An unbounded
/// cumulative curve (equity) keeps the min/max envelope so drawdowns survive. Keyed
/// on the tap name — `exposure` is the only bounded level tap today.
fn reduce_for_tap(tap: &str) -> ReduceKind {
    if tap == "exposure" {
        ReduceKind::Mean
    } else {
        ReduceKind::MinMax
    }
}

/// Serve-time decimation on the aligned `ChartData` (#108). Partition the shared
/// `xs` into at most `buckets` contiguous index ranges; per non-empty bucket emit the
/// bucket's first (and, if it spans >1 index, last) timestamp as shared spine slots,
/// and reduce each series per its [`ReduceKind`] (#111): a `MinMax` series emits min
/// then max (the envelope — equity drawdowns survive), a `Mean` series emits the
/// per-bucket mean in both slots (the net level — a bounded exposure shows its
/// duty-cycle instead of a -1..+1 band). An all-null bucket emits null. `meta` passes
/// through unchanged. Deterministic (C1). Full data stays on disk; only the served
/// page is thinned. No-op when `xs.len() <= 2 * buckets`.
fn decimate(data: ChartData, buckets: usize) -> ChartData {
    let buckets = buckets.max(1);
    let n = data.xs.len();
    if n <= 2 * buckets {
        return data;
    }
    let ChartData { xs, series, meta } = data;

    // Bucket index bounds (lo, hi_exclusive, two_slots) + the decimated shared spine.
    // xs is sorted+deduped (strictly increasing) -> boundary timestamps are strictly
    // increasing across and within buckets, so the spine stays monotonic for uPlot.
    let mut bounds: Vec<(usize, usize, bool)> = Vec::with_capacity(buckets);
    let mut out_xs: Vec<i64> = Vec::with_capacity(2 * buckets);
    for b in 0..buckets {
        let lo = b * n / buckets;
        let hi = (b + 1) * n / buckets;
        if lo >= hi {
            continue;
        }
        let two = hi - lo > 1;
        out_xs.push(xs[lo]);
        if two {
            out_xs.push(xs[hi - 1]);
        }
        bounds.push((lo, hi, two));
    }

    let out_series: Vec<Series> = series
        .into_iter()
        .map(|s| {
            let mut points: Vec<Option<f64>> = Vec::with_capacity(out_xs.len());
            for &(lo, hi, two) in &bounds {
                let (first, second) = match s.reduce {
                    ReduceKind::MinMax => {
                        // envelope: min at the first slot, max at the second.
                        let mut mn = f64::INFINITY;
                        let mut mx = f64::NEG_INFINITY;
                        let mut any = false;
                        for v in s.points[lo..hi].iter().flatten() {
                            any = true;
                            if *v < mn {
                                mn = *v;
                            }
                            if *v > mx {
                                mx = *v;
                            }
                        }
                        if any { (Some(mn), Some(mx)) } else { (None, None) }
                    }
                    ReduceKind::Mean => {
                        // net level: the per-bucket mean written to both slots (a flat
                        // step), so a bounded high-flip series shows its duty-cycle
                        // instead of a -1..+1 band (#111).
                        let mut sum = 0.0;
                        let mut cnt = 0u32;
                        for v in s.points[lo..hi].iter().flatten() {
                            sum += *v;
                            cnt += 1;
                        }
                        let m = if cnt > 0 { Some(sum / cnt as f64) } else { None };
                        (m, m)
                    }
                };
                points.push(first);
                if two {
                    points.push(second);
                }
            }
            Series { name: s.name, y_scale_id: s.y_scale_id, points, reduce: s.reduce }
        })
        .collect();

    ChartData { xs: out_xs, series: out_series, meta }
}

/// Build the serve-ready `ChartData` from a run's read-back traces by the spec-§6
/// 3-step union-spine alignment — no tap privileged, no point dropped:
/// (1) xs = the sorted, deduped union of every tap's timestamps;
/// (2) synthesize an empty-payload spine over xs and pass ALL taps (via
///     `ColumnarTrace::to_rows`, which yields uniformly-f64 rows) as symmetric sides
///     of `join_on_ts`, so no side row is dropped and none occupies the privileged
///     `JoinedRow.spine`;
/// (3) flatten each (tap, column) to a `Series` of `Option<f64>` over xs.
fn build_chart_data(name: &str, traces: RunTraces) -> ChartData {
    let mut xs: Vec<i64> = traces.taps.iter().flat_map(|t| t.ts.iter().copied()).collect();
    xs.sort_unstable();
    xs.dedup();

    let spine: Vec<(Timestamp, Vec<Scalar>)> = xs.iter().map(|&t| (Timestamp(t), Vec::new())).collect();
    let tap_rows: Vec<Vec<(Timestamp, Vec<Scalar>)>> = traces.taps.iter().map(|t| t.to_rows()).collect();
    let sides: Vec<&[(Timestamp, Vec<Scalar>)]> = tap_rows.iter().map(|r| r.as_slice()).collect();
    let joined: Vec<JoinedRow> = join_on_ts(&spine, &sides);

    let mut series: Vec<Series> = Vec::new();
    for (i, tap) in traces.taps.iter().enumerate() {
        for c in 0..tap.columns.len() {
            let name = if tap.columns.len() == 1 { tap.tap.clone() } else { format!("{}[{c}]", tap.tap) };
            let y_scale_id = format!("y_{}", series.len());
            let points: Vec<Option<f64>> =
                joined.iter().map(|r| r.sides[i].as_ref().map(|row| row[c].as_f64())).collect();
            series.push(Series { name, y_scale_id, points, reduce: reduce_for_tap(&tap.tap) });
        }
    }

    let m = &traces.manifest;
    let meta = ChartMeta {
        kind: "run".to_string(),
        name: name.to_string(),
        commit: m.commit.clone(),
        window: (m.window.0.0, m.window.1.0),
        broker: m.broker.clone(),
        seed: m.seed,
        taps: traces.taps.iter().map(|t| t.tap.clone()).collect(),
        members: None,
        params: m.params.iter().map(|(k, v)| (k.clone(), render_value(v))).collect(),
    };
    ChartData { xs, series, meta }
}

/// One member's contribution to the comparison build: its key (the future series
/// name) paired with the chosen tap's drained `(ts, row)` pairs.
type MemberRows = (String, Vec<(Timestamp, Vec<Scalar>)>);

/// Build the comparison `ChartData` for a family: one `Series` per member (the
/// chosen `tap`'s column), labelled by `member.key`, ALL sharing ONE `y_scale_id`
/// (the members measure one identical quantity, so a shared scale is what makes
/// them comparable — unlike the single-run overlay, whose series are different
/// taps). Aligned on the union-ts spine via the same `join_on_ts` build_chart_data
/// uses. `Err` if NO member carries `tap` (refuse-don't-guess).
fn build_comparison_chart_data(
    name: &str,
    members: &[FamilyMember],
    tap: &str,
) -> Result<ChartData, String> {
    let mut member_rows: Vec<MemberRows> = Vec::new();
    for m in members {
        if let Some(t) = m.traces.taps.iter().find(|t| t.tap == tap) {
            member_rows.push((m.key.clone(), t.to_rows()));
        }
    }
    if member_rows.is_empty() {
        return Err(format!("no family member has a tap named '{tap}'"));
    }

    let mut xs: Vec<i64> =
        member_rows.iter().flat_map(|(_, r)| r.iter().map(|(t, _)| t.0)).collect();
    xs.sort_unstable();
    xs.dedup();

    let spine: Vec<(Timestamp, Vec<Scalar>)> =
        xs.iter().map(|&t| (Timestamp(t), Vec::new())).collect();
    let sides: Vec<&[(Timestamp, Vec<Scalar>)]> =
        member_rows.iter().map(|(_, r)| r.as_slice()).collect();
    let joined: Vec<JoinedRow> = join_on_ts(&spine, &sides);

    // One shared y-scale across all member series (same quantity).
    let y_scale_id = format!("y_cmp_{tap}");
    let mut series: Vec<Series> = Vec::new();
    for (i, (key, _)) in member_rows.iter().enumerate() {
        // Project column 0 — the doc's "chosen tap's column" (singular). The
        // comparison taps in scope (equity / exposure) are single-column `f64`.
        // A future multi-column tap selection would need a column index here.
        let points: Vec<Option<f64>> =
            joined.iter().map(|r| r.sides[i].as_ref().map(|row| row[0].as_f64())).collect();
        series.push(Series { name: key.clone(), y_scale_id: y_scale_id.clone(), points, reduce: reduce_for_tap(tap) });
    }
    // member_rows is non-empty here (checked above) => members is non-empty, so
    // members[0] is safe. commit/broker ARE shared across a family (one frozen
    // artifact, one broker profile), but the window is NOT: a walk-forward family's
    // members are disjoint OOS windows (commit 4c64feb), so the family window is the
    // SPAN across all members — (min from, max to). For sweep/MC, whose members
    // share one window, the span collapses to that shared window, so this is the one
    // correct reading for all three kinds.
    let m = &members[0].traces.manifest;
    let window = (
        members.iter().map(|fm| fm.traces.manifest.window.0.0).min().unwrap(),
        members.iter().map(|fm| fm.traces.manifest.window.1.0).max().unwrap(),
    );
    let meta = ChartMeta {
        kind: "family".to_string(),
        name: name.to_string(),
        commit: m.commit.clone(),
        window,
        broker: m.broker.clone(),
        seed: m.seed,
        taps: vec![tap.to_string()],
        members: Some(members.len()),
        params: Vec::new(),
    };
    Ok(ChartData { xs, series, meta })
}

/// Restrict a single-run `ChartData` to the one series named `tap`. `Err` if the
/// run has no such tap (refuse-don't-guess). Used by the `--tap` flag on the
/// single-run chart path; without `--tap` the single-run page is unchanged.
fn filter_to_tap(data: ChartData, tap: &str) -> Result<ChartData, String> {
    if !data.series.iter().any(|s| s.name == tap) {
        // List the valid taps so the user can correct a typo (#131) — refuse, but help.
        let available: Vec<&str> = data.series.iter().map(|s| s.name.as_str()).collect();
        return Err(format!("run has no tap named '{tap}' (available: {})", available.join(", ")));
    }
    let series: Vec<Series> = data.series.into_iter().filter(|s| s.name == tap).collect();
    let mut meta = data.meta;
    meta.taps = vec![tap.to_string()];
    Ok(ChartData { xs: data.xs, series, meta })
}

/// `aura chart <name> [--tap <t>] [--panels]`: classify the name and render. A
/// single run charts all its taps (or the one `--tap` selects); a family overlays
/// one tap (default `equity`) across its members; an unknown name is a runtime error
/// (stderr + exit 1), never a panic.
fn emit_chart(name: &str, tap: Option<&str>, mode: ChartMode, env: &project::Env) {
    let store = env.trace_store();
    render_chart_by_kind(name, store.name_kind(name), tap, mode, env, &store);
}

/// Render (or refuse) a chart for a name whose [`NameKind`] is already known —
/// the shared tail `emit_chart` calls directly for the original name, and
/// [`resolve_campaign_name`]'s `NotFound` arm re-enters with the resolved
/// trace-store handle's own kind (never the un-resolved name), so the
/// handle-charted path (Run/Family) is exercised exactly once either way.
fn render_chart_by_kind(
    name: &str, kind: NameKind, tap: Option<&str>, mode: ChartMode, env: &project::Env,
    store: &aura_registry::TraceStore,
) {
    match kind {
        NameKind::Run => {
            let traces = match store.read(name) {
                Ok(t) => t,
                Err(e) => {
                    eprintln!("aura: {e}");
                    std::process::exit(1);
                }
            };
            let mut data = build_chart_data(name, traces);
            if let Some(t) = tap {
                data = match filter_to_tap(data, t) {
                    Ok(d) => d,
                    Err(e) => {
                        eprintln!("aura: {e}");
                        std::process::exit(1);
                    }
                };
            }
            let data = decimate(data, CHART_DECIMATE_BUCKETS);
            print!("{}", render::render_chart_html(&data, mode));
        }
        NameKind::Family => {
            let members = match store.read_family(name) {
                Ok(m) => m,
                Err(e) => {
                    eprintln!("aura: {e}");
                    std::process::exit(1);
                }
            };
            let data = match build_comparison_chart_data(name, &members, tap.unwrap_or("equity")) {
                Ok(d) => d,
                Err(e) => {
                    eprintln!("aura: {e}");
                    std::process::exit(1);
                }
            };
            let data = decimate(data, CHART_DECIMATE_BUCKETS);
            print!("{}", render::render_chart_html(&data, mode));
        }
        NameKind::NotFound => match resolve_campaign_name(name, env) {
            NameResolution::Unique(handle) => {
                let resolved_kind = store.name_kind(&handle);
                render_chart_by_kind(&handle, resolved_kind, tap, mode, env, store);
            }
            NameResolution::Ambiguous(handles) => {
                eprintln!(
                    "aura: the campaign name '{name}' names {} recorded runs ({}) — \
                     chart one of these handles directly",
                    handles.len(),
                    handles.join(", "),
                );
                std::process::exit(1);
            }
            NameResolution::None => {
                eprintln!(
                    "aura: no recorded run or family '{name}' under runs/traces \
                     (check the handle a sweep/walk-forward/campaign run printed for a typo — \
                     re-running with this handle as `--trace` will not create it)"
                );
                std::process::exit(1);
            }
        },
    }
}

/// The outcome of resolving a `chart <NAME>` argument against the recorded
/// campaign runs (#238): `NAME` is not a trace-store handle, but it may be the
/// `--trace <NAME>` the user chose when the family was produced, which lands
/// only in the campaign document's `name` field (`put_campaign`), never in the
/// trace-store layout itself.
enum NameResolution {
    /// Exactly one recorded campaign run's stored document carries `name ==
    /// NAME` and a persisted `trace_name` — the trace-store handle to chart.
    Unique(String),
    /// More than one recorded run's document carries `name == NAME` — refuse
    /// rather than silently pick one; the candidate handles, in the campaign
    /// store's file (append) order, so the refusal is deterministic.
    Ambiguous(Vec<String>),
    /// No recorded run's document carries `name == NAME` — NAME is genuinely
    /// unknown, not a chosen campaign name either.
    None,
}

/// Resolve a `chart` argument against `env.registry()`'s campaign-run records:
/// keep every record whose `trace_name` is `Some` (a family was actually
/// persisted for it) AND whose stored campaign document (`get_campaign`)
/// parses with `name == name`. Malformed/missing campaign documents are
/// skipped rather than propagated — a resolution failure downgrades to "not a
/// campaign name", not a hard error, since the trace-store NotFound message is
/// still an honest fallback.
fn resolve_campaign_name(name: &str, env: &project::Env) -> NameResolution {
    let registry = env.registry();
    let records = match registry.load_campaign_runs() {
        Ok(r) => r,
        Err(_) => return NameResolution::None,
    };
    let mut candidates = Vec::new();
    for record in &records {
        let Some(trace_name) = &record.trace_name else { continue };
        let Ok(Some(doc_json)) = registry.get_campaign(&record.campaign) else { continue };
        let Ok(doc) = aura_research::parse_campaign(&doc_json) else { continue };
        if doc.name == name {
            candidates.push(trace_name.clone());
        }
    }
    match candidates.len() {
        0 => NameResolution::None,
        1 => NameResolution::Unique(candidates.remove(0)),
        _ => NameResolution::Ambiguous(candidates),
    }
}

/// What `--real` parsing yields: the synthetic default, or a real symbol + an
/// optional window (parsed, not yet opened). Pure, so the grammar is unit-testable.
#[derive(Debug, Clone, PartialEq)]
enum DataChoice {
    Synthetic,
    Real { symbol: String, from_ms: Option<i64>, to_ms: Option<i64> },
}

/// The source provider threaded into the family builders: synthetic built-in
/// streams, or real M1 close bars from the data-server archive. Replaces the
/// hardcoded `VecSource` so a member's source, pip, window, and roller sizes come
/// from one place (Fork B/D/F).
///
/// `Synthetic` denotes a *consumer-dependent* built-in stream, not one fixed
/// series: the full-window consumers (`full_window` / `run_sources`, used by
/// sweep / MC) draw the 18-bar `showcase_prices()`, while the windowed consumers
/// (`windowed_sources` / `wf_window_sizes`, used by walk-forward) draw the 60-bar
/// `walkforward_prices()` so the `(24,12,12)`-bar roller fits its span. The two
/// faces never reach one consumer (a family is either full-window or windowed), so
/// the split is invisible per call site but real across the type — read both
/// family builders to see it whole.
enum DataSource {
    Synthetic,
    Real {
        server: std::sync::Arc<data_server::DataServer>,
        symbol: String,
        from_ms: Option<i64>,
        to_ms: Option<i64>,
        pip: f64,
    },
}

/// No-local-data refusal — stderr + exit(1).
fn no_real_data(symbol: &str, env: &project::Env) -> ! {
    eprintln!("aura: no local data for symbol '{symbol}' at {}", env.data_path());
    std::process::exit(1)
}

/// Empty-in-window refusal — stderr + exit(1). Distinct from `no_real_data`:
/// used only inside `probe_window`, whose callers have already proven the
/// symbol present via `has_symbol` — an empty probe result there is a fact
/// about the requested `--from`/`--to` window, not about symbol absence, so
/// it must not reuse the symbol-absence message (#242).
fn no_data_in_window(symbol: &str, from_ms: Option<i64>, to_ms: Option<i64>, env: &project::Env) -> ! {
    let bound = |b: Option<i64>| b.map_or_else(|| "unbounded".to_string(), |v| v.to_string());
    eprintln!(
        "aura: no data for symbol '{symbol}' in the requested window [{}, {}] at {}",
        bound(from_ms),
        bound(to_ms),
        env.data_path()
    );
    std::process::exit(1)
}

/// Resolve the per-instrument pip from the recorded geometry sidecar, or refuse
/// (stderr + exit 1) when the symbol has no recorded geometry — the single home of
/// the guessed-pip refusal, shared by `open_real_source` and `from_choice`. Reads
/// the symbol's geometry metadata (not bar data) before the run source opens, so an
/// instrument with no recorded geometry refuses without a guessed pip; the pip is
/// the provider's recorded value, honest by construction.
fn pip_or_refuse(
    server: &std::sync::Arc<data_server::DataServer>, symbol: &str, env: &project::Env,
) -> f64 {
    match aura_ingest::instrument_geometry(server, symbol) {
        Some(geo) => geo.pip_size,
        None => {
            eprintln!(
                "aura: no recorded geometry for symbol '{symbol}' at {} — \
                 refusing to run a real instrument with a guessed pip",
                env.data_path()
            );
            std::process::exit(1);
        }
    }
}

/// Probe the full data window: open a single-pass probe source through the
/// shared opener, drain it for the first/last timestamp, and return
/// `(first, last)`. Refuses (via `no_data_in_window`) when the symbol/window
/// yields no source or no bars — callers have already proven the symbol
/// present, so an empty result here is a window fact, not symbol absence.
/// Shared by `open_real_source` (which needs the manifest
/// window from a probe separate from the run sources) and
/// `DataSource::full_window`. The probe drains the CLOSE column regardless of
/// the strategy's binding: the bounds are field-independent (every archived bar
/// carries all six fields at one timestamp), and file-level absence is
/// field-independent too.
fn probe_window(
    server: &std::sync::Arc<data_server::DataServer>,
    symbol: &str,
    from_ms: Option<i64>,
    to_ms: Option<i64>,
    env: &project::Env,
) -> (Timestamp, Timestamp) {
    aura_ingest::archive_extent(server, std::path::Path::new(&env.data_path()), symbol, from_ms, to_ms)
        .unwrap_or_else(|| no_data_in_window(symbol, from_ms, to_ms, env))
}

/// Open the real M1 sources for a recorded symbol over an optional window — one
/// source per resolved binding column, in canonical order — returning the run
/// sources paired with the manifest `window` and the per-instrument `pip_size`.
/// Single home of the real-source construction the single-run handlers share — the
/// sidecar-pip lookup, the `DataServer` `has_symbol` refusal, the probe-window pass, and
/// the run-source open (each refusal an stderr + exit 1). Pre-data refusals keep the
/// pip honest by construction. Used by `resolve_run_data`.
fn open_real_source(
    symbol: &str,
    from_ms: Option<i64>,
    to_ms: Option<i64>,
    env: &project::Env,
    fields: &[aura_ingest::M1Field],
) -> (Vec<Box<dyn aura_engine::Source>>, (Timestamp, Timestamp), f64) {
    // Per-instrument pip from the recorded sidecar; resolved BEFORE bar-data access
    // so an instrument with no geometry refuses without touching the archive.
    let server = std::sync::Arc::new(data_server::DataServer::new(env.data_path()));
    let pip = pip_or_refuse(&server, symbol, env);
    if !server.has_symbol(symbol) {
        no_real_data(symbol, env);
    }
    // Manifest window: drain a separate probe (single-pass Source) for first/last ts.
    let window = probe_window(&server, symbol, from_ms, to_ms, env);
    let sources = aura_ingest::open_columns(&server, symbol, from_ms, to_ms, fields)
        .unwrap_or_else(|| no_real_data(symbol, env));
    (sources, window, pip)
}

impl DataSource {
    /// Build a provider from a parsed choice, or refuse (stderr + exit 1) on a symbol
    /// with no recorded geometry / absent data — both BEFORE any member runs (Fork C/G),
    /// via the same `pip_or_refuse` / `no_real_data` helpers `open_real_source` uses.
    fn from_choice(choice: DataChoice, env: &project::Env) -> DataSource {
        match choice {
            DataChoice::Synthetic => DataSource::Synthetic,
            DataChoice::Real { symbol, from_ms, to_ms } => {
                let server = std::sync::Arc::new(data_server::DataServer::new(env.data_path()));
                let pip = pip_or_refuse(&server, &symbol, env);
                if !server.has_symbol(&symbol) {
                    no_real_data(&symbol, env);
                }
                DataSource::Real { server, symbol, from_ms, to_ms, pip }
            }
        }
    }

    fn pip_size(&self) -> f64 {
        match self {
            DataSource::Synthetic => SYNTHETIC_PIP_SIZE,
            DataSource::Real { pip, .. } => *pip,
        }
    }

    /// The full run window, probed once. Synthetic: the showcase span. Real:
    /// `probe_window` drains a separate single-pass probe source for first/last ts
    /// (the same helper `open_real_source` uses for its manifest window).
    fn full_window(&self, env: &project::Env) -> (Timestamp, Timestamp) {
        match self {
            DataSource::Synthetic => {
                let s: Vec<Box<dyn aura_engine::Source>> = vec![Box::new(VecSource::new(showcase_prices()))];
                window_of(&s).expect("non-empty showcase stream")
            }
            DataSource::Real { server, symbol, from_ms, to_ms, .. } => {
                probe_window(server, symbol, *from_ms, *to_ms, env)
            }
        }
    }

    /// The full walk-forward span. Synthetic draws the 60-bar `walkforward_prices`
    /// span — NOT `showcase_prices` (which `full_window` uses): walk-forward is a
    /// *windowed* consumer whose roller `(24,12,12)` needs 36 bars, so it uses the
    /// longer built-in stream (byte-unchanged from the retired pre-`DataSource`
    /// `walkforward_family`, which derived its span the same way). Real: the same
    /// probed `--from..--to` window as `full_window`.
    fn wf_full_span(&self, env: &project::Env) -> (Timestamp, Timestamp) {
        match self {
            DataSource::Synthetic => {
                let s: Vec<Box<dyn aura_engine::Source>> = vec![Box::new(VecSource::new(walkforward_prices()))];
                window_of(&s).expect("non-empty walkforward stream")
            }
            DataSource::Real { server, symbol, from_ms, to_ms, .. } => {
                probe_window(server, symbol, *from_ms, *to_ms, env)
            }
        }
    }

    /// A fresh full-window source set per member (single-pass): the synthetic
    /// showcase close stream, or one real source per resolved binding column
    /// in canonical order (callers guard the synthetic arm to `{close}`).
    fn run_sources(&self, env: &project::Env, fields: &[aura_ingest::M1Field]) -> Vec<Box<dyn aura_engine::Source>> {
        match self {
            DataSource::Synthetic => vec![Box::new(VecSource::new(showcase_prices()))],
            DataSource::Real { server, symbol, from_ms, to_ms, .. } => {
                aura_ingest::open_columns(server, symbol, *from_ms, *to_ms, fields)
                    .unwrap_or_else(|| no_real_data(symbol, env))
            }
        }
    }

    /// A fresh windowed source set for an IS/OOS sub-window (walk-forward).
    /// Synthetic: `walkforward_window_source`. Real: one source per resolved
    /// binding column over the ns-native window.
    fn windowed_sources(
        &self, from: Timestamp, to: Timestamp, env: &project::Env, fields: &[aura_ingest::M1Field],
    ) -> Vec<Box<dyn aura_engine::Source>> {
        match self {
            DataSource::Synthetic => vec![Box::new(walkforward_window_source(from, to))],
            DataSource::Real { server, symbol, .. } => {
                aura_ingest::open_columns_window(server, symbol, Some(from), Some(to), fields)
                    .unwrap_or_else(|| no_real_data(symbol, env))
            }
        }
    }

    /// WindowRoller sizes per data kind (Fork F): bar-index for synthetic (24/12/12
    /// over the 60-bar span), calendar-ns for real.
    fn wf_window_sizes(&self) -> (i64, i64, i64) {
        match self {
            DataSource::Synthetic => (24, 12, 12),
            DataSource::Real { .. } => (WF_REAL_IS_NS, WF_REAL_OOS_NS, WF_REAL_STEP_NS),
        }
    }

}

/// The honest broker label for the dual-tap r-sma harness: it runs a RiskExecutor
/// branch alongside the SimBroker, so the plain "sim-optimal" label would under-report
/// it (#132). Shared by the single run and the sweep so the two cannot drift.
fn r_sma_broker_label(pip_size: f64) -> String {
    format!("sim-optimal+risk-executor(pip_size={pip_size})")
}

/// In-sample winner-selection objective for walk-forward (cycle 0077). `Argmax` is
/// the bare-best pick deflated for trials (#144, the default); `Plateau` argmaxes
/// the neighbourhood-smoothed surface instead (opt-in via `--select`).
#[derive(Clone, Copy)]
enum Selection {
    Argmax,
    Plateau(PlateauMode),
}

/// Parse a `--select` token: `argmax` | `plateau:mean` | `plateau:worst`. Unknown
/// tokens are a usage error (the caller maps `Err(())` to exit 2).
fn parse_select(s: &str) -> Result<Selection, ()> {
    match s {
        "argmax" => Ok(Selection::Argmax),
        "plateau:mean" => Ok(Selection::Plateau(PlateauMode::Mean)),
        "plateau:worst" => Ok(Selection::Plateau(PlateauMode::Worst)),
        _ => Err(()),
    }
}

/// Map the CLI `--select` value to the research selection rule the campaign
/// document carries.
fn select_rule_of(sel: Selection) -> aura_research::SelectRule {
    match sel {
        Selection::Argmax => aura_research::SelectRule::Argmax,
        Selection::Plateau(PlateauMode::Mean) => aura_research::SelectRule::PlateauMean,
        Selection::Plateau(PlateauMode::Worst) => aura_research::SelectRule::PlateauWorst,
    }
}

/// Parse a comma-separated list of `T` (each item parsed via `FromStr`), rejecting
/// any item that fails to parse — the shared validator for the r-sma grid flags
/// (#137). The caller folds the `Err` into the subcommand `usage()` so a malformed
/// `--fast 2,x` is the strict usage path, not a downstream panic. `str::split(',')`
/// always yields at least one item, and an empty item (`""`, or a trailing comma)
/// fails its per-item `parse`, so a non-empty result needs no separate empty-list
/// check — the empty/blank inputs are rejected by the per-item parse itself.
fn parse_csv_list<T: std::str::FromStr>(s: &str) -> Result<Vec<T>, ()> {
    let items: Vec<&str> = s.split(',').collect();
    let mut out = Vec::with_capacity(items.len());
    for item in items {
        out.push(item.parse::<T>().map_err(|_| ())?);
    }
    Ok(out)
}

/// Render a family-member stdout line: the assigned `family_id` plus the embedded
/// `RunReport`. The report is emitted in its own declaration key order (manifest
/// leads with `commit`, C18) so the line is byte-identical to the stored
/// `families.jsonl`. `serde_json::json!` would route the report through a
/// `serde_json::Value` and re-alphabetize the manifest keys (broker-first),
/// diverging from the store — hence the report is spliced in pre-serialized (#99).
fn family_member_line(id: &str, report: &RunReport) -> String {
    format!(
        r#"{{"family_id":{},"report":{}}}"#,
        serde_json::to_string(id).expect("a string id always serializes"),
        report.to_json()
    )
}

/// Monte-Carlo variant of [`family_member_line`]: the per-draw line also carries the
/// realization `seed` (between `family_id` and `report`) — the shape `run_blueprint_mc` prints.
fn mc_member_line(id: &str, seed: u64, report: &RunReport) -> String {
    format!(
        r#"{{"family_id":{},"seed":{},"report":{}}}"#,
        serde_json::to_string(id).expect("a string id always serializes"),
        seed,
        report.to_json()
    )
}

/// Resolve the in-sample winner under the chosen selection objective. `Argmax`
/// defers to the trials-deflation pick (#144). `Plateau` argmaxes the smoothed grid
/// surface — it needs the grid lattice, so a sweep with no lattice (a future random
/// walk-forward producer) is refused rather than silently argmaxed. The metric is
/// always known at the call sites, so a metric error is unreachable (`expect`); the
/// only fallible outcome is the plateau-without-lattice refusal, returned as
/// `Err(message)` for the caller to print and exit 2.
fn select_winner(
    family: &SweepFamily, metric: &str, select: Selection, lattice: Option<&[usize]>,
) -> Result<(SweepPoint, FamilySelection), String> {
    match select {
        Selection::Argmax => Ok(optimize_deflated(
            family, metric, DEFLATION_N_RESAMPLES, DEFLATION_BLOCK_LEN, DEFLATION_SEED,
        ).expect("walk-forward metrics are known")),
        Selection::Plateau(mode) => match lattice {
            Some(lens) => Ok(optimize_plateau(family, lens, metric, mode)
                .expect("walk-forward metrics are known")),
            None => Err(
                "--select plateau requires a grid sweep; a random sweep has no parameter lattice"
                    .to_string(),
            ),
        },
    }
}

/// Pool every OOS window's per-trade R series into one flat vector, in roll order
/// (window order, then within-window trade order). Windows with no `r` block
/// contribute nothing. The single home of the pooling-in-roll-order semantics —
/// both the walk-forward `oos_r` summary and the `mc` R-bootstrap reduce this.
fn pooled_oos_net_trade_rs(result: &WalkForwardResult) -> Vec<f64> {
    result
        .windows
        .iter()
        .flat_map(|w| w.run.oos_report.metrics.r.as_ref().map(|r| r.net_trade_rs.clone()).unwrap_or_default())
        .collect()
}

/// 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);
    let pooled_rs = pooled_oos_net_trade_rs(result);
    let mut obj = serde_json::json!({
        "windows": result.windows.len(),
        "stitched_total_pips": total,
        "param_stability": param_stability(result),
    });
    if result.windows.iter().any(|w| w.run.oos_report.metrics.r.is_some()) {
        // RMetrics serializes its scalar fields (net_trade_rs is serde-skipped, so the
        // oos_r block is the clean R-metric summary of the pooled series).
        obj["oos_r"] = serde_json::to_value(r_metrics_from_rs(&pooled_rs))
            .expect("RMetrics serializes");
    }
    serde_json::json!({ "walkforward": obj }).to_string()
}

/// The walk-forward summary line reconstructed from the recorded per-window OOS
/// reports (the campaign path's `WalkForward` `StageFamily.reports`) rather than a
/// live `WalkForwardResult`. `stitched_total_pips` = the per-window `total_pips`
/// summed left-to-right in roll order (the engine `stitch` folds each segment's
/// final cumulative value, and a window's OOS segment ends at its `total_pips`);
/// `param_stability` reduces each IS-refit axis in `axes` over the per-window
/// chosen params (read from each report's `manifest.params` via
/// [`campaign_run::raw_matches_wrapped`] — blueprint axes are recorded wrapped
/// (e.g. `sma_signal.fast.length`) by the strategy's own param space, while
/// `stop_length`/`stop_k` ride the risk regime unwrapped, so an exact-name match
/// would miss the wrapped ones) through the same `MetricStats::from_values`; the
/// `oos_r` block pools the per-window `net_trade_rs` through `r_metrics_from_rs`.
/// Canonical JSON (C14).
///
/// `axes` carries the invocation's raw axis names in argv order, followed by the
/// stop columns when a regime is bound (#220 — no axis name is hardcoded here).
/// Order matters: the committed exact-grade anchor pins
/// `param_stability[0].mean` = the first axis's refit mean.
fn walkforward_summary_json_from_reports(reports: &[RunReport], axes: &[String]) -> String {
    let total: f64 = reports.iter().map(|r| r.metrics.total_pips).sum();
    // Coercion to `f64` is decided per-value at runtime by the `Scalar` variant
    // match below (i64 axes cast value-as-f64, the f64 axis passes through),
    // exactly as `param_stability`'s per-`ScalarKind` coerce does — the axis
    // list carries no per-axis type flag.
    let param_stability: Vec<aura_engine::MetricStats> = axes
        .iter()
        .map(|axis| {
            let vals: Vec<f64> = reports
                .iter()
                .map(|r| {
                    let (_, v) = r
                        .manifest
                        .params
                        .iter()
                        .find(|(name, _)| campaign_run::raw_matches_wrapped(axis, name))
                        .expect("each walk-forward window records its chosen axis");
                    match v {
                        Scalar::I64(i) => *i as f64,
                        Scalar::F64(f) => *f,
                        Scalar::Bool(b) => *b as i64 as f64,
                        Scalar::Timestamp(_) => {
                            unreachable!("a timestamp is a structural axis, never a param knob")
                        }
                    }
                })
                .collect();
            aura_engine::MetricStats::from_values(&vals)
        })
        .collect();
    let pooled_rs: Vec<f64> = reports
        .iter()
        .flat_map(|r| r.metrics.r.as_ref().map(|m| m.net_trade_rs.clone()).unwrap_or_default())
        .collect();
    let mut obj = serde_json::json!({
        "windows": reports.len(),
        "stitched_total_pips": total,
        "param_stability": param_stability,
    });
    if reports.iter().any(|r| r.metrics.r.is_some()) {
        obj["oos_r"] = serde_json::to_value(r_metrics_from_rs(&pooled_rs))
            .expect("RMetrics serializes");
    }
    serde_json::json!({ "walkforward": obj }).to_string()
}

/// The cross-instrument generalization line: the chosen metric, instrument count,
/// worst-case floor, sign-agreement count, and the per-instrument breakdown. Canonical
/// JSON (C14), mirroring `walkforward_summary_json`'s `{"generalize": obj}` shape.
fn generalize_json(agg: &Generalization) -> String {
    let per: Vec<serde_json::Value> = agg
        .per_instrument
        .iter()
        .map(|(sym, v)| serde_json::json!([sym, v]))
        .collect();
    let obj = serde_json::json!({
        "metric": agg.selection_metric,
        "n_instruments": agg.n_instruments,
        "worst_case": agg.worst_case,
        "sign_agreement": agg.sign_agreement,
        "per_instrument": per,
    });
    serde_json::json!({ "generalize": obj }).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 synthetic path uses (the firewall mapping to
/// `DataServer::stream_m1_windowed` is the real-data path; synthetic stays in-memory,
/// mirroring `run_blueprint_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 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,
            "bias_sign_flips": agg.bias_sign_flips,
        }
    })
    .to_string()
}

/// Render an `RBootstrap` as one canonical JSON line (`MetricStats` serializes; the
/// scalar fields are spliced in), mirroring `mc_aggregate_json`.
fn mc_r_bootstrap_json(b: &RBootstrap) -> String {
    serde_json::json!({
        "mc_r_bootstrap": {
            "n_trades": b.n_trades,
            "block_len": b.block_len,
            "n_resamples": b.n_resamples,
            "e_r": b.e_r,
            "prob_le_zero": b.prob_le_zero,
        }
    })
    .to_string()
}

/// `aura runs families`: one header line per stored family (id, kind, member
/// count), in first-seen store order.
fn runs_families(env: &project::Env) {
    let reg = env.registry();
    let members = match reg.load_family_members() {
        Ok(m) => m,
        Err(e) => {
            eprintln!("aura: {e}");
            std::process::exit(1);
        }
    };
    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); an unknown metric is a usage error (stderr + exit 2).
fn runs_family(id: &str, rank: Option<&str>, env: &project::Env) {
    let reg = env.registry();
    let members = match reg.load_family_members() {
        Ok(m) => m,
        Err(e) => {
            eprintln!("aura: {e}");
            std::process::exit(1);
        }
    };
    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());
        if let Some(sel) = &report.manifest.selection {
            match sel.mode {
                // `deflated_score` is `None` only on an Argmax record with no
                // deflation run (report.rs); guard it, symmetric with the
                // plateau branch, so a from-disk record cannot panic here. When
                // present (the sole producer always stamps it), the bytes are
                // unchanged.
                SelectionMode::Argmax => if let Some(deflated) = sel.deflated_score {
                    match sel.overfit_probability {
                        Some(p) => println!("  deflated={deflated:.4} P(overfit)={p:.4}"),
                        None => println!("  deflated={deflated:.4}"),
                    }
                },
                SelectionMode::PlateauMean | SelectionMode::PlateauWorst => {
                    let label = if matches!(sel.mode, SelectionMode::PlateauMean) { "mean" } else { "worst" };
                    if let (Some(score), Some(n)) = (sel.neighbourhood_score, sel.n_neighbours) {
                        println!("  plateau({label})={score:.4} over {n} cells");
                    }
                }
            }
        }
    }
}

/// The outcome of reproducing one persisted family: per member, whether its re-run
/// metrics are bit-identical to the stored metrics (C1).
struct ReproduceReport {
    outcomes: Vec<(String, bool)>,
}

/// Reconstruct a member's bootstrap point from its recorded named params — the inverse
/// of `zip_params(space, point)`. Walks the reloaded signal's `param_space` in order
/// (deterministic for the same blueprint) and reads each knob's value from the manifest.
fn point_from_params(space: &[ParamSpec], params: &[(String, Scalar)]) -> Vec<Cell> {
    space
        .iter()
        .map(|ps| {
            params
                .iter()
                .find(|(n, _)| n == &ps.name)
                .map(|(_, s)| s.cell())
                .unwrap_or_else(|| {
                    // A manifest missing a param the reloaded space expects is
                    // corrupted-on-disk data — exit cleanly (`aura:` + exit 1) like
                    // every other persisted-data failure on the reproduce path, not
                    // a panic.
                    eprintln!("aura: manifest is missing param {}", ps.name);
                    std::process::exit(1);
                })
        })
        .collect()
}

/// Re-derive the `StopRule` a member was minted under from its manifest params
/// (`stop_length`/`stop_k`, or `stop_period_minutes`/`stop_length`/`stop_k` for
/// the #262 timescale-matched variant, stamped by `run_blueprint_member` — the
/// stop rides the risk regime OUTSIDE the wrapped param_space, so
/// `point_from_params` cannot recover it). `stop_period_minutes` present
/// re-derives `VolTf`; otherwise falls back to `Vol`, and to the default
/// vol-stop regime when the manifest carries no stop knobs at all (pre-#233
/// members), mirroring `stop_rule_for_regime`'s `None` arm (campaign_run.rs)
/// for the same one-directional widening `point_from_params` already applies
/// to missing manifest params.
fn stop_rule_from_params(params: &[(String, Scalar)]) -> StopRule {
    let period_minutes =
        params.iter().find(|(n, _)| n == "stop_period_minutes").map(|(_, s)| s.as_i64());
    let length = params.iter().find(|(n, _)| n == "stop_length").map(|(_, s)| s.as_i64());
    let k = params.iter().find(|(n, _)| n == "stop_k").map(|(_, s)| s.as_f64());
    match (period_minutes, length, k) {
        (Some(period_minutes), Some(length), Some(k)) => {
            StopRule::VolTf { period_minutes, length, k }
        }
        (None, Some(length), Some(k)) => StopRule::Vol { length, k },
        _ => StopRule::Vol { length: R_SMA_STOP_LENGTH, k: R_SMA_STOP_K },
    }
}

/// Re-derive the cost model a member was minted under from its manifest params
/// (`cost[k].<knob>`, stamped by `run_blueprint_member`) — the #233 stop-regime
/// pattern: like the stop, the cost model rides OUTSIDE the wrapped
/// param_space, so `point_from_params` cannot recover it. Components are read
/// in index order; the builder knob name discriminates the variant (each
/// shipped cost node has exactly one, distinctly named knob). No `cost[0].*`
/// param = the empty (gross) model — every pre-cost member widens to it,
/// mirroring `stop_rule_from_params`'s default arm.
fn cost_specs_from_params(params: &[(String, Scalar)]) -> Vec<aura_research::CostSpec> {
    let mut specs = Vec::new();
    for k in 0.. {
        let prefix = format!("cost[{k}].");
        let Some((knob, v)) = params
            .iter()
            .find_map(|(n, s)| n.strip_prefix(&prefix).map(|rest| (rest, *s)))
        else {
            break;
        };
        specs.push(match knob {
            "cost_per_trade" => aura_research::CostSpec::Constant {
                cost_per_trade: aura_research::CostValue::Scalar(v.as_f64()),
            },
            "slip_vol_mult" => aura_research::CostSpec::VolSlippage {
                slip_vol_mult: aura_research::CostValue::Scalar(v.as_f64()),
            },
            "carry_per_cycle" => aura_research::CostSpec::Carry {
                carry_per_cycle: aura_research::CostValue::Scalar(v.as_f64()),
            },
            other => {
                // Corrupted-on-disk manifest data — the reproduce path's clean
                // exit register (`aura:` + exit 1), like point_from_params.
                eprintln!("aura: manifest cost param cost[{k}].{other} names no cost component");
                std::process::exit(1);
            }
        });
    }
    specs
}

/// Look up a persisted family by id, or exit 1 (unknown id / registry load failure) —
/// the single place `reproduce_family` and `reproduce_family_in` resolve a family, so
/// the two exit-1 error phrasings can't drift out of sync between the call sites.
fn load_family_or_exit(reg: &Registry, id: &str) -> Family {
    let members = reg.load_family_members().unwrap_or_else(|e| {
        eprintln!("aura: {e}");
        std::process::exit(1);
    });
    group_families(members).into_iter().find(|f| f.id == id).unwrap_or_else(|| {
        // reproduce is an action, not a lookup: an unknown id is a hard error (distinct
        // from `runs family <id>`'s treat-as-empty exit 0).
        eprintln!("aura: no such family '{id}'");
        std::process::exit(1);
    })
}

/// Re-derive every member of a persisted sweep family from the content-addressed store
/// and compare to the stored result, against an explicit registry (testable seam).
fn reproduce_family_in(
    reg: &Registry,
    id: &str,
    data: &DataSource,
    env: &project::Env,
) -> ReproduceReport {
    let family = load_family_or_exit(reg, id);
    let pip = data.pip_size();
    let mut outcomes = Vec::new();
    for member in &family.members {
        let stored = &member.report;
        let hash = stored.manifest.topology_hash.clone().unwrap_or_else(|| {
            eprintln!("aura: family member has no topology_hash; not a generated run");
            std::process::exit(1);
        });
        let doc = reg
            .get_blueprint(&hash)
            .unwrap_or_else(|e| {
                eprintln!("aura: {e}");
                std::process::exit(1);
            })
            .unwrap_or_else(|| {
                eprintln!("aura: blueprint {hash} missing from store");
                std::process::exit(1);
            });
        // The #246 override set (silent variant): re-derived from the RECORDED
        // manifest param names, against a raw probe space + raw strategy load — a
        // name matching neither space is simply not an override (falls through to
        // `point_from_params`'s existing missing-knob refusal below), unlike the
        // sweep boundary's `override_paths`, which errors on it.
        let recorded: Vec<String> =
            stored.manifest.params.iter().map(|(n, _)| n.clone()).collect();
        let raw_space = blueprint_axis_probe(&doc, env).param_space();
        let raw_signal = blueprint_from_json(&doc, &|t| env.resolve(t)).unwrap_or_else(|e| {
            eprintln!("aura: stored blueprint {hash} does not parse: {e:?}");
            std::process::exit(1);
        });
        let overrides = wrapped_bound_overrides_of(&recorded, &raw_space, &raw_signal);
        // Reload the stored blueprint per use: a Composite is !Clone, and both the
        // param-space probe (below) and the re-run each consume one. The doc was
        // canonical-serialized at store time, so every reload is infallible. Every
        // reload re-opens the same override set derived above, so the recorded
        // param names resolve against the space they were minted under (#246).
        let reload = || {
            reopen_all(
                blueprint_from_json(&doc, &|t| env.resolve(t)).unwrap_or_else(|e| {
                    eprintln!("aura: stored blueprint {hash} does not parse: {e:?}");
                    std::process::exit(1);
                }),
                &overrides,
            )
        };
        // The param_space of the WRAPPED signal — its knobs carry the `r_sma`
        // wrapper's `sma_signal.` node-path prefix, exactly the names the manifest
        // recorded at write time. Mirrors `blueprint_sweep_family`'s probe so the
        // reproduce-side space name-matches the stored params (raw `signal.param_space()`
        // would drop the prefix and `point_from_params` could not find the knobs).
        let (tx_eq, _) = mpsc::channel();
        let (tx_ex, _) = mpsc::channel();
        let (tx_r, _) = mpsc::channel();
        let (tx_req, _) = mpsc::channel();
        let stop = stop_rule_from_params(&stored.manifest.params);
        // The member's cost model, re-derived from its manifest (the #233
        // stop pattern): a costed family re-runs under the exact components it
        // was minted with, so `net_expectancy_r` reproduces bit-identically.
        let cost = cost_specs_from_params(&stored.manifest.params);
        // The member's binding, re-derived from the stored blueprint's own
        // input roles (name defaults — family manifests carry no overrides).
        let binding = binding::resolve_binding(&hash, reload().input_roles(), &BTreeMap::new())
            .unwrap_or_else(|m| {
                eprintln!("aura: {m}");
                std::process::exit(1);
            });
        if matches!(data, DataSource::Synthetic) && !binding.close_only() {
            eprintln!("aura: {}", binding::synthetic_refusal(&hash, &binding));
            std::process::exit(1);
        }
        let space = wrap_r(reload(), tx_eq, tx_ex, tx_r, tx_req, stop, true, SYNTHETIC_PIP_SIZE, &binding, None).param_space();
        let point = point_from_params(&space, &stored.manifest.params);
        // A MonteCarlo member carries no tuning params (the params-join is empty), so its
        // reproduce line would print a BLANK member label; the seed IS its realization
        // identity, so surface `seed=<N>` instead. Sweep / walk-forward members echo their
        // tuning params (the params-join), unchanged.
        let label = match family.kind {
            FamilyKind::MonteCarlo => format!("seed={}", stored.manifest.seed),
            _ => stored
                .manifest
                .params
                .iter()
                .map(|(n, v)| format!("{n}={}", render_value(v)))
                .collect::<Vec<_>>()
                .join(", "),
        };
        // Realization-aware: a MonteCarlo member ran over a seed-driven synthetic walk,
        // not the showcase — reconstruct it from manifest.seed so the re-run matches (C1).
        // The seed is manifest-carried and identical across realizations; only the sources
        // and window vary by kind.
        let seed = stored.manifest.seed;
        let (sources, member_window) = match family.kind {
            FamilyKind::MonteCarlo => {
                let s = synthetic_walk_sources(seed);
                let w = window_of(&s).expect("non-empty synthetic walk");
                (s, w)
            }
            FamilyKind::WalkForward => {
                // each member is one OOS window: rebuild its windowed slice from the
                // stored window bounds; the winner params come from the shared
                // manifest->cells recovery below (as Sweep members do).
                let (from, to) = stored.manifest.window;
                let s = data.windowed_sources(from, to, env, &binding.columns());
                let w = window_of(&s).expect("non-empty OOS window");
                (s, w)
            }
            // Sweep / plain-run members: a Real source reopens the exact per-member
            // window the manifest recorded, the same `windowed_sources` loader
            // WalkForward uses above (and `CliMemberRunner` uses at mint time, #229)
            // — never a fresh full-archive probe, which need not match the window the
            // family was minted over. Synthetic keeps the pre-#229 full-window path
            // (byte-identical: `data.full_window` is a pure, no-IO computation).
            _ => match data {
                DataSource::Real { .. } => {
                    let (from, to) = stored.manifest.window;
                    (data.windowed_sources(from, to, env, &binding.columns()), (from, to))
                }
                DataSource::Synthetic => (data.run_sources(env, &binding.columns()), data.full_window(env)),
            },
        };
        let rerun = run_blueprint_member(
            reload(),
            &point,
            &space,
            sources,
            member_window,
            seed,
            pip,
            &hash,
            env,
            stop,
            &binding,
            &cost,
            // The re-run specs come from the stamp and are scalar by
            // construction (`cost_specs_from_params`), so the instrument is
            // inert; the fallback is never resolved against a map.
            stored.manifest.instrument.as_deref().unwrap_or(""),
        );
        outcomes.push((label, rerun.metrics == stored.metrics));
    }
    ReproduceReport { outcomes }
}

/// `aura reproduce <family-id>`: re-derive a persisted sweep family from the
/// content-addressed blueprint store and verify each member reproduces bit-identically
/// (C18 "re-derives full results on demand"). The data source is reconstructed from
/// the family's own manifest (#229) — a hardcoded `DataSource::Synthetic` here would
/// re-derive a real-data family over the wrong stream; see `reproduce_family_in`'s
/// per-member window loader for how the reconstructed source is actually used.
fn reproduce_family(id: &str, env: &project::Env) {
    let reg = env.registry();
    let family = load_family_or_exit(&reg, id);
    // Reconstruct the DataSource the family was minted over: `None` instrument
    // (every synthetic-family member, pre-#229 lines) stays the built-in synthetic
    // stream; a real instrument reopens the same local archive `--real` runs use,
    // via the same named-data refusal (exit 1) on a missing sidecar/archive.
    let data = match family.members.first().and_then(|m| m.report.manifest.instrument.clone()) {
        None => DataSource::Synthetic,
        Some(symbol) => {
            DataSource::from_choice(DataChoice::Real { symbol, from_ms: None, to_ms: None }, env)
        }
    };
    let rep = reproduce_family_in(&reg, id, &data, env);
    let total = rep.outcomes.len();
    let ok = rep.outcomes.iter().filter(|(_, b)| *b).count();
    for (label, identical) in &rep.outcomes {
        let verdict = if *identical { "bit-identical" } else { "DIVERGED" };
        println!("{id}  member {label}  reproduced: {verdict}");
    }
    println!("reproduced {ok}/{total} members bit-identically");
    if ok != total {
        std::process::exit(1);
    }
}

// --- r-sma harness (the SMA-cross signal scored in R) --------------------

/// The r-sma vol-stop EWMA length (cycles). Single source for the `StopRule::Vol`
/// the blueprint embeds and the `stop` param the manifest records — kept honest by one
/// constant instead of a hand-synced literal across the function boundary.
pub(crate) const R_SMA_STOP_LENGTH: i64 = 3;

/// The r-sma vol-stop multiplier (1R = `k`·σ). Single source for the embedded
/// `StopRule::Vol` and its manifest record, like [`R_SMA_STOP_LENGTH`].
pub(crate) const R_SMA_STOP_K: f64 = 2.0;

/// Short-horizon realized-range window for vol-scaled slippage. Deliberately
/// distinct from `R_SMA_STOP_LENGTH` (3): scaling slippage by the stop's own
/// vol would collapse cost-in-R to a constant (spec 0082). Short enough to warm
/// within the synthetic smoke fixture so the run path exercises non-zero slippage.
const SLIP_VOL_LENGTH: i64 = 5;

/// The optional cost leg of the R scaffolding (#234): the resolved cost-node
/// builders (from `campaign_run::cost_nodes_for`, fully bound — they add no
/// open param, so the wrapped `param_space` is cost-invariant) plus the two
/// recording channels — the aggregate 3-field cost record (`tx_cost`, the
/// `summarize_r` join input) and the `net_r_equity` curve (`tx_net`, recorded
/// only in `!reduce` trace mode).
struct CostLeg {
    nodes: Vec<PrimitiveBuilder>,
    tx_cost: mpsc::Sender<(Timestamp, Vec<Scalar>)>,
    tx_net: mpsc::Sender<(Timestamp, Vec<Scalar>)>,
}

/// 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.
#[derive(Debug)]
enum RunData {
    Synthetic,
    Real { symbol: String, from: Option<i64>, to: Option<i64> },
}

/// A rise-fall-rise synthetic stream for the r-sma 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 r_sma_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()
}

/// Interned `col[i]` recorder-port names, built once. The `GraphBuilder::input` API
/// wants `&'static str`; interning here (instead of `format!(...).leak()` per field)
/// means reusing the r-sma harness in a sweep / Monte-Carlo loop reuses these
/// strings rather than leaking 14 fresh ones per build (#132).
static COL_PORTS: LazyLock<Vec<String>> =
    LazyLock::new(|| (0..PM_FIELD_NAMES.len()).map(|i| format!("col[{i}]")).collect());

/// SHA256 (hex) of a canonical (#164) blueprint JSON string — the content id (#158).
/// The single hashing primitive, shared by [`topology_hash`] (from a live `Composite`)
/// and the op-script `graph introspect --content-id` path (`crate::content_id`), so the
/// two surfaces agree by construction over the same canonical bytes. Research-side
/// (aura-cli), off the frozen engine (invariant 8).
fn content_id(canonical_json: &str) -> String {
    aura_research::content_id_of(canonical_json)
}

/// SHA256 (hex) of the canonical (#164, no-trailing-newline) serialization of a
/// signal blueprint — the run's `topology_hash` (#158).
fn topology_hash(signal: &Composite) -> String {
    content_id(&blueprint_to_json(signal).expect("a buildable signal serializes"))
}

/// Wrap a `signal` composite (a roles→`bias` leg) in the R run
/// scaffolding: pip broker, the per-tap recorders, the vol-stop RiskExecutor, the
/// r_equity / cost legs. The signal is nested; its root roles come from the
/// resolved `binding` (one per entry, canonical column order — the same order the
/// callers open real columns in, so role i receives source i), and the binding's
/// guaranteed close entry always feeds the broker/executor pair. A serialized
/// signal loaded via `blueprint_from_json` runs through exactly the scaffolding
/// the Rust-built signal does.
#[allow(clippy::type_complexity, clippy::too_many_arguments)]
fn wrap_r(
    signal: Composite,
    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>)>,
    stop: StopRule,
    reduce: bool,
    pip_size: f64,
    binding: &binding::ResolvedBinding,
    cost: Option<CostLeg>,
) -> Composite {
    let mut g = GraphBuilder::new("r_sma");
    // The strategy signal → Bias, nested as a serializable roles→`bias` leg.
    let sig = g.add(BlueprintNode::Composite(signal));
    // pip branch (verbatim from the retired `sample_blueprint_with_sinks`, #159).
    let broker = g.add(SimBroker::builder(pip_size));
    // R branch: bias + price → RiskExecutor(vol_stop) → dense R-record. The stop is a
    // fixed `StopRule` — the pinned default constants, or an arbitrary per-regime rule
    // the caller resolved.
    // In `reduce` mode the per-cycle taps fold online: SeriesReducer folds the eq/ex
    // f64 series to one summary row, GatedRecorder retains only the gated R rows — the
    // O(cycles)→O(trades) memory win. The raw `Recorder`s (and the r_equity tap) are the
    // `--trace` path, where the full per-cycle series is persisted.
    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(stop, 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))
    };
    // Root roles come from the resolved binding, one per entry in canonical
    // column order (the C4 merge tie-break order the callers open columns in).
    // The binding guarantees a close entry, which always feeds the broker and
    // the executor — "price" below is their node-schema PORT name, not a role
    // name. For a single-`price` signal this declares exactly the old weld:
    // one role, targets [sig, broker, exec] (feed calls append targets).
    let mut close_handle = None;
    for entry in binding.entries() {
        let role = g.source_role(&entry.role, binding::column_kind(entry.column));
        if entry.feeds_signal {
            // `NodeHandle::input` takes a `&'static str` port name (the fluent
            // `GraphBuilder`'s authoring-time contract, builder.rs) but the
            // binding's role names are dynamic (loaded from a blueprint at
            // runtime). Vocabulary names use their static form; only an
            // override-renamed role leaks its name (once per graph build,
            // never per-tick).
            let port: &'static str = binding::static_role_name(&entry.role)
                .unwrap_or_else(|| Box::leak(entry.role.clone().into_boxed_str()));
            g.feed(role, [sig.input(port)]);
        }
        if entry.column == aura_ingest::M1Field::Close && close_handle.is_none() {
            close_handle = Some(role);
        }
    }
    let close = close_handle.expect("ResolvedBinding guarantees a close entry");
    g.feed(close, [broker.input("price"), exec.input("price")]);
    g.connect(sig.output("bias"), broker.input("exposure"));
    g.connect(sig.output("bias"), ex.input("col[0]"));
    g.connect(sig.output("bias"), exec.input("bias"));
    g.connect(broker.output("equity"), eq.input("col[0]"));
    for (i, field) in PM_FIELD_NAMES.iter().enumerate() {
        g.connect(exec.output(field), rrec.input(COL_PORTS[i].as_str()));
    }
    if !reduce {
        // 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));
        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]"));
    }
    // The optional cost leg (#234 — the #221-deleted wiring rebuilt as a
    // campaign-documented feature): a `cost_graph` fed by the executor's four
    // geometry outputs, its aggregate record recorded co-temporally with the R
    // record (the `summarize_r` positional-join input), and — in `!reduce`
    // trace mode — the LinComb(4) net-equity curve into `tx_net`. Absent leg
    // (`None`) == today's graph, byte-identical.
    if let Some(leg) = cost {
        // Components taking a `volatility` extra input (the vol_slippage
        // shape), discovered from each builder's own schema past the geometry
        // prefix — no second vocabulary of "which node needs the proxy".
        let vol_slots: Vec<usize> = leg
            .nodes
            .iter()
            .enumerate()
            .filter(|(_, n)| {
                n.schema().inputs[GEOMETRY_WIDTH..].iter().any(|p| p.name == "volatility")
            })
            .map(|(k, _)| k)
            .collect();
        // The short-horizon realized-range vol proxy, fed from the close role,
        // shared by every vol-scaled component (the #221-deleted arm, verbatim
        // wiring; built in BOTH modes — the reduce-mode member run charges
        // slippage too).
        let vol_range = if vol_slots.is_empty() {
            None
        } else {
            let vhi = g.add(RollingMax::builder().named("slip_vol_hi").bind("length", Scalar::i64(SLIP_VOL_LENGTH)));
            let vlo = g.add(RollingMin::builder().named("slip_vol_lo").bind("length", Scalar::i64(SLIP_VOL_LENGTH)));
            let vrange = g.add(Sub::builder().named("slip_vol_range"));
            g.connect(vhi.output("value"), vrange.input("lhs"));
            g.connect(vlo.output("value"), vrange.input("rhs"));
            g.feed(close, [vhi.input("series"), vlo.input("series")]);
            Some(vrange)
        };
        // One cost_graph composite fans the shared PM-geometry into the
        // components and sums their per-field charges (C10).
        let cg = g.add(cost_graph(leg.nodes));
        g.connect(exec.output("closed_this_cycle"), cg.input("closed"));
        g.connect(exec.output("open"), cg.input("open"));
        g.connect(exec.output("entry_price"), cg.input("entry_price"));
        g.connect(exec.output("stop_price"), cg.input("stop_price"));
        for k in vol_slots {
            let vrange = vol_range.as_ref().expect("proxy is built whenever a vol slot exists");
            g.connect(vrange.output("value"), cg.input(cost_port(k, "volatility")));
        }
        if reduce {
            // The aggregate cost record, gated on the SAME closed flag as the R
            // GatedRecorder and flushed once at finalize — so the cost rows stay
            // positionally 1:1 with the gated R rows (`summarize_r`'s join). The
            // gate rides as an APPENDED col 3: cols 0..2 keep the aura-analysis
            // `cost_col` contract (cost_in_r = 0, open_cost_in_r = 2).
            let crec = g.add(GatedRecorder::builder(
                vec![ScalarKind::F64, ScalarKind::F64, ScalarKind::F64, ScalarKind::Bool],
                3,
                Firing::Any,
                leg.tx_cost,
            ));
            g.connect(cg.output("cost_in_r"), crec.input("col[0]"));
            g.connect(cg.output("cum_cost_in_r"), crec.input("col[1]"));
            g.connect(cg.output("open_cost_in_r"), crec.input("col[2]"));
            g.connect(exec.output("closed_this_cycle"), crec.input("col[3]"));
        } else {
            // The full per-cycle aggregate cost record (col 0 per-close, col 2
            // window-end — what summarize_r folds on the trace path).
            let crec = g.add(Recorder::builder(
                vec![ScalarKind::F64, ScalarKind::F64, ScalarKind::F64],
                Firing::Any,
                leg.tx_cost,
            ));
            g.connect(cg.output("cost_in_r"), crec.input("col[0]"));
            g.connect(cg.output("cum_cost_in_r"), crec.input("col[1]"));
            g.connect(cg.output("open_cost_in_r"), crec.input("col[2]"));
            // net_r_equity = cum_realized_r + unrealized_r − Σcum_cost_in_r
            // − Σopen_cost_in_r (the #221-deleted LinComb(4), weights 1,1,-1,-1).
            let net_eq = g.add(
                LinComb::builder(4)
                    .bind("weights[0]", Scalar::f64(1.0))
                    .bind("weights[1]", Scalar::f64(1.0))
                    .bind("weights[2]", Scalar::f64(-1.0))
                    .bind("weights[3]", Scalar::f64(-1.0)),
            );
            g.connect(exec.output("cum_realized_r"), net_eq.input("term[0]"));
            g.connect(exec.output("unrealized_r"), net_eq.input("term[1]"));
            g.connect(cg.output("cum_cost_in_r"), net_eq.input("term[2]"));
            g.connect(cg.output("open_cost_in_r"), net_eq.input("term[3]"));
            let net_rec = g.add(Recorder::builder(vec![ScalarKind::F64], Firing::Any, leg.tx_net));
            g.connect(net_eq.output("value"), net_rec.input("col[0]"));
        }
    }
    g.build().expect("r_sma wiring resolves")
}

/// Pair each opened source with its role name from the resolved binding, forming
/// the keyed supply `run_bound` resolves by name. `sources` are opened in
/// `binding.columns()` order (== `binding.entries()` order), so entry `i`'s role
/// names source `i` — the one place open-order and role-order meet, made explicit
/// so `bind_sources` verifies the wiring↔supply role match by name (#275).
fn key_supply(
    binding: &binding::ResolvedBinding,
    sources: Vec<Box<dyn aura_engine::Source>>,
) -> Vec<(String, Box<dyn aura_engine::Source>)> {
    binding
        .entries()
        .iter()
        .map(|e| e.role.clone())
        .zip(sources)
        .collect()
}

/// Resolve a `RunData` selector to the `(sources, window, pip_size)` triple the
/// run paths feed to the harness: the built-in synthetic R stream (close-only —
/// the caller guards the binding shape), or the lazily-streamed real sources of
/// the binding's resolved columns (with the sidecar pip + probed window).
/// Single definition used by `run_signal_r`.
#[allow(clippy::type_complexity)]
fn resolve_run_data(
    data: &RunData,
    env: &project::Env,
    binding: &binding::ResolvedBinding,
) -> (
    Vec<Box<dyn aura_engine::Source>>,
    (Timestamp, Timestamp),
    f64,
) {
    match data {
        RunData::Synthetic => {
            let sources: Vec<Box<dyn aura_engine::Source>> =
                vec![Box::new(VecSource::new(r_sma_prices()))];
            let window = window_of(&sources).expect("non-empty synthetic stream");
            (sources, window, SYNTHETIC_PIP_SIZE)
        }
        RunData::Real { symbol, from, to } => {
            open_real_source(symbol, *from, *to, env, &binding.columns())
        }
    }
}

/// Run a signal blueprint through the R scaffolding: hash the signal,
/// wrap it (broker + equity/exposure/R sinks), compile with `params`, bootstrap,
/// run over `data`, and build the RunReport (manifest carries topology_hash).
/// The single construction+run path shared by the `aura run <blueprint.json>` CLI
/// arm and its bit-identical test.
#[allow(clippy::type_complexity)]
fn run_signal_r(
    signal: Composite, params: &[Scalar], data: RunData, seed: u64, env: &project::Env,
) -> RunReport {
    let topo = topology_hash(&signal); // before signal is consumed
    let run_name = signal.name().to_string(); // before signal is consumed by `wrap_r`
    // The default binding (name defaults; `aura run` carries no campaign
    // overrides). Refusals are the established `aura: ` + exit-1 register.
    let binding = binding::resolve_binding(signal.name(), signal.input_roles(), &BTreeMap::new())
        .unwrap_or_else(|m| {
            eprintln!("aura: {m}");
            std::process::exit(1);
        });
    if matches!(data, RunData::Synthetic) && !binding.close_only() {
        eprintln!("aura: {}", binding::synthetic_refusal(signal.name(), &binding));
        std::process::exit(1);
    }
    let names: Vec<String> = signal
        .param_space()
        .iter()
        .map(|p| p.name.clone())
        .collect();
    let defaults = wrapped_bound_defaults(&signal); // read before `signal` is consumed below
    let (tx_eq, rx_eq) = mpsc::channel();
    let (tx_ex, rx_ex) = mpsc::channel();
    let (tx_r, rx_r) = mpsc::channel();
    // The req tap (r_equity recorder) is wired but not persisted on this path; keep the
    // receiver alive so the sink's sends do not fail, but do not drain it.
    let (tx_req, _rx_req) = mpsc::channel();
    let (sources, window, pip_size) = resolve_run_data(&data, env, &binding);
    let wrapped = wrap_r(signal, tx_eq, tx_ex, tx_r, tx_req, StopRule::Vol { length: R_SMA_STOP_LENGTH, k: R_SMA_STOP_K }, false, pip_size, &binding, None);
    let mut flat = wrapped.compile_with_params(params).unwrap_or_else(|e| {
        eprintln!("aura: this blueprint does not compile to a runnable harness: {e:?}");
        std::process::exit(1);
    });
    // Bind each declared tap to a fresh `Recorder` + channel, before bootstrap
    // — `flat.taps` already carries the signal's declared taps hoisted to the
    // root. Dedup is the caller's per `TapBindError::DuplicateBind`'s doc (the
    // engine keeps no cross-call state).
    let mut seen: BTreeSet<String> = BTreeSet::new();
    let mut tap_drains: Vec<(String, ScalarKind, mpsc::Receiver<(Timestamp, Vec<Scalar>)>)> = Vec::new();
    let declared: Vec<aura_engine::FlatTap> = flat.taps.clone();
    for tap in &declared {
        if !seen.insert(tap.name.clone()) {
            eprintln!("aura: {}", aura_engine::TapBindError::DuplicateBind { name: tap.name.clone() });
            std::process::exit(1);
        }
        let kind = flat.signatures[tap.node].output[tap.field].kind;
        let (tx, rx) = mpsc::channel();
        let sig = Recorder::builder(vec![kind], Firing::Any, tx.clone()).schema().clone();
        flat.bind_tap(&tap.name, Box::new(Recorder::new(&[kind], Firing::Any, tx)), sig)
            .expect("declared tap binds (name from flat.taps, kind from its own signature)");
        tap_drains.push((tap.name.clone(), kind, rx));
    }
    let mut h = Harness::bootstrap(flat).expect("valid r-sma harness");
    // `sources` were opened via `resolve_run_data(&data, env, &binding)` against
    // this SAME `binding`, and `key_supply` keys them by that binding's own role
    // names — a `SourceBindError` here can only mean the wiring↔supply pairing
    // this function builds is internally inconsistent, never a user input
    // mistake. Panic like the adjacent bootstrap invariants above, not a
    // process exit dressed as a refusal message.
    h.run_bound(key_supply(&binding, sources))
        .expect("sources opened against `binding` key-match that binding's own roles by construction");
    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 named_params: Vec<(String, Scalar)> =
        names.into_iter().zip(params.iter().copied()).collect();
    let mut manifest = sim_optimal_manifest(named_params, window, seed, pip_size);
    manifest.defaults = defaults;
    manifest.broker = r_sma_broker_label(pip_size);
    manifest.topology_hash = Some(topo);
    manifest.project = env.provenance();
    let mut metrics = summarize(&f64_field(&eq_rows, 0), &f64_field(&ex_rows, 0));
    metrics.r = Some(summarize_r(&r_rows, &[]));
    // Drain + persist each declared tap's series, guarded so a tap-free
    // `aura run` stays byte-identical to today (no `runs/` write at all).
    // `manifest` is built above so the persisted `index.json` carries this
    // run's own provenance.
    if !tap_drains.is_empty() {
        let tap_traces: Vec<ColumnarTrace> = tap_drains
            .into_iter()
            .map(|(name, kind, rx)| {
                let rows: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
                ColumnarTrace::from_rows(&name, &[kind], &rows)
            })
            .collect();
        env.trace_store().write(&run_name, &manifest, &tap_traces).unwrap_or_else(|e| {
            eprintln!("aura: writing tap traces failed: {e}");
            std::process::exit(1);
        });
    }
    RunReport { manifest, metrics }
}

/// #260: the r-sma sugar/MC paths below run with either no cost model (empty
/// `cost` slice) or CLI-flag cost specs (scalar-only — `cost_specs_from_params`
/// only ever wraps `CostValue::Scalar`), so an instrument-keyed map can never
/// originate on these paths and the instrument context is genuinely inert.
/// Named once so every such call site states its intent by reference instead
/// of repeating the justifying comment.
const NO_INSTRUMENT_CONTEXT: &str = "";

/// Run one bootstrapped member of a loaded-signal sweep: the reduce-mode path
/// (`SeriesReducer` folds eq/ex, `GatedRecorder` retains the gated R rows — the
/// O(cycles)→O(trades) fold), shared by the live sweep AND reproduction so a
/// reproduced member re-derives bit-identically (C1). `signal` is a freshly
/// reloaded blueprint (`Composite` is `!Clone`); `point` is the member's bound
/// cells; `space` gives the by-name manifest params; `topo` the shared signal hash.
#[allow(clippy::too_many_arguments)]
fn run_blueprint_member(
    signal: Composite,
    point: &[Cell],
    space: &[ParamSpec],
    sources: Vec<Box<dyn aura_engine::Source>>,
    window: (Timestamp, Timestamp),
    seed: u64,
    pip: f64,
    topo: &str,
    env: &project::Env,
    stop: StopRule,
    binding: &binding::ResolvedBinding,
    cost: &[aura_research::CostSpec],
    instrument: &str,
) -> RunReport {
    let defaults = wrapped_bound_defaults(&signal); // read before `signal` is consumed below
    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();
    // The doc's cost model as an optional wrap leg (#234): empty = no leg,
    // exactly the pre-cost graph. The net curve is a !reduce trace concern;
    // its sender is wired but unread here (the r_equity tap precedent above).
    let (tx_cost, rx_cost) = mpsc::channel();
    let (tx_net, _rx_net) = mpsc::channel();
    let cost_leg = (!cost.is_empty()).then(|| CostLeg {
        nodes: campaign_run::cost_nodes_for(cost, instrument),
        tx_cost,
        tx_net,
    });
    let mut h = wrap_r(signal, tx_eq, tx_ex, tx_r, tx_req, stop, true, pip, binding, cost_leg)
        .bootstrap_with_cells(point)
        .expect("member bootstraps (point kind-checked against param_space)");
    // Called from inside `CliMemberRunner::run_member`'s `catch_unwind`
    // containment (#272 — that impl's doc contract: "never a process exit
    // inside a sweep worker, every refusal a member fault"). A
    // `SourceBindError` here is an internal wiring/supply mismatch, not user
    // input (see the sibling `.expect` two lines above) — panic so the
    // containing `catch_unwind` records it as a per-cell `MemberFault::Panic`
    // instead of `process::exit` aborting the whole campaign.
    h.run_bound(key_supply(binding, sources))
        .expect("sources opened against `binding` key-match that binding's own roles by construction");
    let mut named = zip_params(space, point); // by-name params for the manifest record
    // `match` (not an irrefutable `let`): `StopRule` also has a `Fixed` variant,
    // which stamps no vol knobs. The campaign/single-run paths only pass `Vol`
    // or `VolTf`, so the `Fixed` arm is inert here.
    match stop {
        StopRule::Vol { length, k } => {
            named.push(("stop_length".to_string(), Scalar::i64(length)));
            named.push(("stop_k".to_string(), Scalar::f64(k)));
        }
        StopRule::VolTf { period_minutes, length, k } => {
            named.push(("stop_period_minutes".to_string(), Scalar::i64(period_minutes)));
            named.push(("stop_length".to_string(), Scalar::i64(length)));
            named.push(("stop_k".to_string(), Scalar::f64(k)));
        }
        StopRule::Fixed(_) => {}
    }
    // Stamp the cost model the member ran under, beside the stop knobs (#234,
    // the #233 pattern): one `cost[k].<knob>` param per component, in
    // component order — the knob name discriminates the variant (each shipped
    // cost node has exactly one, distinctly named knob). The name is read off
    // `campaign_run::cost_knob`, the same function `cost_nodes_for` binds
    // through, so the stamp key cannot drift from the bind key: the manifest
    // carries enough to re-derive the exact model later. `reproduce_family_in`
    // reads this stamp back via `cost_specs_from_params` (the #233 stop-regime
    // pattern), so a costed family reproduces net, not gross.
    for (k, spec) in cost.iter().enumerate() {
        let (knob, v) = campaign_run::cost_knob(spec, instrument);
        named.push((format!("cost[{k}].{knob}"), Scalar::f64(v)));
    }
    let mut manifest = sim_optimal_manifest(named, window, seed, pip);
    manifest.defaults = defaults;
    manifest.broker = r_sma_broker_label(pip);
    manifest.topology_hash = Some(topo.to_string());
    manifest.project = env.provenance();
    let r_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_r.try_iter().collect();
    // The member's cost rows (empty when no cost model) join the R reduction:
    // net_expectancy_r diverges from gross by exactly the modelled costs.
    let cost_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_cost.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, &cost_rows));
    RunReport { manifest, metrics: m }
}

/// The exact wrapped probe the loaded-blueprint sweep resolves its axes
/// against: the loaded signal wrapped in the r-sma scaffolding (stop bound,
/// reduce, no cost), taps discarded. `param_space()` on it is the axis
/// namespace `--axis` binds; `.axis()` consumes it to seed a sweep. Single
/// source for the sweep terminal, the MC closed-check, AND `--list-axes`, so
/// the listed names track the swept names by construction (incl. across #159's
/// harness retirement). The reload is infallible under the SAME
/// dispatch-boundary contract the callers already rely on: the doc is
/// `blueprint_from_json`-validated at the `["sweep", ..]` / `["mc", ..]`
/// boundary before this runs, so a malformed doc has already exited 2 and
/// never reaches the `.expect`.
fn blueprint_axis_probe(doc: &str, env: &project::Env) -> Composite {
    blueprint_axis_probe_reopened(doc, env, &[])
}

/// The axis probe with a #246 override set re-opened on the strategy BEFORE
/// wrapping — probe and per-member reloads must re-open identically so points
/// resolve against one space. The empty-set form is byte-equal to the old
/// probe (mc/run closed-checks and the open `--list-axes` lines read that).
fn blueprint_axis_probe_reopened(doc: &str, env: &project::Env, overrides: &[String]) -> Composite {
    let signal = blueprint_from_json(doc, &|t| env.resolve(t))
        .expect("doc parse-validated at the dispatch boundary; reload is infallible");
    let signal = reopen_all(signal, overrides);
    // The PROBE binding is lenient (unresolvable roles fall back to close),
    // like the probe's synthetic pip: the wrap is built for its param_space
    // only, never run over data — strict resolution lives on the run paths.
    let probe = binding::probe_binding(signal.input_roles());
    let (tx_eq, _) = mpsc::channel();
    let (tx_ex, _) = mpsc::channel();
    let (tx_r, _) = mpsc::channel();
    let (tx_req, _) = mpsc::channel();
    wrap_r(signal, tx_eq, tx_ex, tx_r, tx_req, StopRule::Vol { length: R_SMA_STOP_LENGTH, k: R_SMA_STOP_K }, true, SYNTHETIC_PIP_SIZE, &probe, None)
}

/// The WRAPPED-coordinate BOUND param name set of `signal` (#246): every
/// `bound_param_space()` entry prefixed with `<signal.name()>.` — the same
/// coordinate `param_space()`'s OPEN entries live in on the wrapped graph.
/// Extracted because `wrapped_bound_overrides_of`, `override_paths`, and
/// `validate_and_register_axes` each independently rebuilt this exact set
/// (the rule-of-three the neighbouring doc comment itself cites).
fn wrapped_bound_names(signal: &Composite) -> HashSet<String> {
    let prefix = format!("{}.", signal.name());
    signal
        .bound_param_space()
        .into_iter()
        .map(|b| format!("{prefix}{}", b.name))
        .collect()
}

/// The override subset of `names` (#246): every WRAPPED-coordinate name
/// missing the un-reopened wrapped OPEN space but naming a BOUND param of the
/// strategy — returned in STRATEGY coordinates (the wrap prefix
/// `<signal.name()>.` stripped) for `Composite::reopen`. Names matching
/// neither space are skipped here; the caller decides whether that is an
/// error (sweep boundary) or falls through to its existing resolution
/// errors. The silent variant `reproduce_family_in` uses over the RECORDED
/// manifest param names (the sweep boundary uses the stricter
/// `override_paths`, which errors on an unmatched name instead). Contrast
/// `campaign_run::raw_bound_overrides_of`, whose `names` are already RAW
/// (a campaign document's own namespace, #203) and needs no such stripping.
fn wrapped_bound_overrides_of(
    names: &[String],
    open_space: &[ParamSpec],
    signal: &Composite,
) -> Vec<String> {
    let open: HashSet<&str> = open_space.iter().map(|p| p.name.as_str()).collect();
    let prefix = format!("{}.", signal.name());
    let bound = wrapped_bound_names(signal);
    names
        .iter()
        .filter(|n| !open.contains(n.as_str()) && bound.contains(*n))
        .map(|n| n[prefix.len()..].to_string())
        .collect()
}

/// The sweep-boundary variant (#246): like `wrapped_bound_overrides_of`, but
/// an axis matching NEITHER the open nor the bound space is the error — the
/// honest replacement of the retired "fully bound; nothing to sweep" refusal.
fn override_paths(
    axes: &[(String, Vec<Scalar>)],
    open_space: &[ParamSpec],
    signal: &Composite,
) -> Result<Vec<String>, String> {
    let open: HashSet<&str> = open_space.iter().map(|p| p.name.as_str()).collect();
    let prefix = format!("{}.", signal.name());
    let bound = wrapped_bound_names(signal);
    let mut overrides = Vec::new();
    for (name, _) in axes {
        if open.contains(name.as_str()) {
            continue;
        }
        if bound.contains(name) {
            overrides.push(name[prefix.len()..].to_string());
        } else {
            return Err(format!(
                "axis {name}: names no param of this blueprint (open or bound) — \
                 see `aura sweep <bp> --list-axes`"
            ));
        }
    }
    Ok(overrides)
}

/// Renders a [`BindError`] as one-line prose in `override_paths`' sibling
/// register above (#247) — never the raw Rust `Debug` struct name
/// (`KindMismatch { .. }` / `MissingKnob("..")`), the two variants the sweep
/// terminal actually raises past `override_paths`' own pre-flight (which
/// already rejects an unresolvable axis name as prose before either call
/// site below ever reaches the terminal). `UnknownKnob` already carries a
/// fully-prosed message string (wrapped from `override_paths`' own
/// `Result<_, String>`) — unwrapped here rather than Debug-framed (#269), so
/// the walkforward path's rejection reaches stderr as bare prose too.
fn render_bind_error(e: &BindError) -> String {
    match e {
        BindError::MissingKnob(name) => format!(
            "axis {name}: an open param with no axis and no bound default — \
             bind it with `--axis {name}=<value>` — see `aura sweep <bp> --list-axes`"
        ),
        BindError::KindMismatch { knob, expected, got } => format!(
            "axis {knob}: expected {expected:?}, supplied {got:?} — \
             see `aura sweep <bp> --list-axes`"
        ),
        BindError::UnknownKnob(msg) => msg.clone(),
        BindError::DuplicateBinding(name) => format!(
            "axis {name}: bound twice — each param takes exactly one axis — \
             see `aura sweep <bp> --list-axes`"
        ),
        BindError::EmptyAxis(name) => format!(
            "axis {name}: supplies no values — give at least one, e.g. `--axis {name}=2,4`"
        ),
        BindError::EmptyRange(name) => format!(
            "axis {name}: the named range is empty — give it at least one value"
        ),
        // A blueprint defect, not an axis usage error: the point passed name
        // resolution but its bootstrap failed. Prose frame with the compile
        // detail explicitly labelled as internal — per-variant prose for
        // CompileError belongs to the graph-build surface, not this boundary.
        BindError::Compile(e) => format!(
            "the resolved axis point failed to bootstrap — the blueprint is \
             defective at this point, re-validate it with `aura graph build` \
             (internal detail: {e:?})"
        ),
    }
}

/// Apply a validated override set to a freshly loaded strategy (#246).
/// Infallible by contract: the set was derived against this same document at
/// the family boundary.
fn reopen_all(signal: Composite, overrides: &[String]) -> Composite {
    overrides.iter().fold(signal, |s, p| {
        s.reopen(p).expect("override set validated at the family boundary")
    })
}

/// `aura sweep <blueprint.json> --list-axes`: one `<name>:<kind>` line per open
/// sweepable knob, in `param_space()` order (byte-identical to before #246),
/// followed by one `<name>:<kind> default=<value>` line per BOUND param —
/// every bound param IS a sweepable default, re-openable by naming it as an
/// `--axis` (#246), on EVERY blueprint shape (fully open, partially open, or
/// fully closed) — not only the fully-closed case: `--axis`/`override_paths`
/// accepts a bound name regardless of how many knobs happen to be open
/// alongside it, so the discovery surface must list it regardless too. Names
/// are exactly what `--axis` binds.
fn list_blueprint_axes(doc: &str, env: &project::Env) {
    let space = blueprint_axis_probe(doc, env).param_space();
    for p in &space {
        println!("{}:{:?}", p.name, p.kind); // ScalarKind Debug -> I64/F64/Bool/Timestamp
    }
    let signal = blueprint_from_json(doc, &|t| env.resolve(t))
        .expect("doc parse-validated at the dispatch boundary");
    let bp = signal.name().to_string();
    for b in signal.bound_param_space() {
        println!("{bp}.{}:{:?} default={}", b.name, b.kind, render_value(&b.value));
    }
}

/// Sweep a serialized signal `doc` over user-named param-space axes. Structurally it
/// keeps the shape of the retired `r_sma_sweep_family` demo builder (#159), with three
/// deviations. (1) The signal source is
/// `wrap_r(blueprint_from_json(doc))` — a loaded blueprint, not the Rust-built
/// r-sma graph. (2) The signal is RE-loaded from `doc` per member (a `Composite` is
/// `!Clone`, so the throwaway param-space probe and each grid point each reload). (3)
/// The axes are taken verbatim BY NAME (not the four suffix-resolved r-sma knobs):
/// each `(name, vals)` is fed straight to the `SweepBinder`, so an unknown name or a
/// kind mismatch surfaces as the sweep terminal's [`BindError`], rendered to a message
/// string — a named error, never a panic. An axis naming a bound param re-opens it
/// (#246: bound value = default); an axis matching neither space is refused by
/// `override_paths` before any run. Every member manifest carries the shared
/// `topology_hash` of the loaded signal; reduce-mode fold, identical to the retired
/// mirror's default (no-trace) arm.
fn blueprint_sweep_family(
    doc: &str,
    axes: &[(String, Vec<Scalar>)],
    data: &DataSource,
    env: &project::Env,
) -> Result<SweepFamily, String> {
    // Identity + binding read the AUTHORED doc, raw (no override re-open):
    // topology and the resolved role plan are properties of the document, not
    // of any one sweep's axis choices.
    let probe_signal = blueprint_from_json(doc, &|t| env.resolve(t))
        .expect("doc parse-validated at the dispatch boundary; reload is infallible");
    let topo = topology_hash(&probe_signal);
    // Strict binding resolution (name defaults — the verb path carries no
    // campaign overrides): the family's open plan and wrap plan in one value.
    let binding = binding::resolve_binding(probe_signal.name(), probe_signal.input_roles(), &BTreeMap::new())?;
    if matches!(data, DataSource::Synthetic) && !binding.close_only() {
        return Err(binding::synthetic_refusal(probe_signal.name(), &binding));
    }
    let pip = data.pip_size();
    let window = data.full_window(env);
    // The un-reopened wrapped OPEN space (#246): derives the override set (which
    // named axes re-open a bound param) before the real, reopened probe is built —
    // probe and per-member reloads must re-open identically so points resolve
    // against one space. A name matching neither space is the error here.
    let raw_space = blueprint_axis_probe(doc, env).param_space();
    let overrides = override_paths(axes, &raw_space, &probe_signal)?;
    let probe = blueprint_axis_probe_reopened(doc, env, &overrides);
    let space = probe.param_space();
    // The doc is parse-validated at the dispatch boundary (with file-path context),
    // so every reload here is infallible: the builder has a single error contract —
    // the `BindError` returned by the sweep terminal — and no hidden process exit.
    // Member reloads re-open the SAME override set derived above, so every member
    // resolves its axes against the identical (reopened) param space the probe used.
    let reload = |d: &str| {
        reopen_all(
            blueprint_from_json(d, &|t| env.resolve(t))
                .expect("doc parse-validated at the dispatch boundary; reload is infallible"),
            &overrides,
        )
    };
    // seed the named axes verbatim: the first via Composite::axis (consumes the probe),
    // the rest via SweepBinder::axis. resolve_axes name- and kind-checks them at the
    // sweep terminal, so an UnknownKnob / KindMismatch is returned, not panicked.
    let mut iter = axes.iter();
    let (first_name, first_vals) = iter.next().expect("a blueprint sweep declares >= 1 axis");
    let mut binder = probe.axis(first_name, first_vals.clone());
    for (n, vals) in iter {
        binder = binder.axis(n, vals.clone());
    }
    binder
        .sweep(|point| {
            // fresh per-member graph (Composite is !Clone, reload per member) run through
            // the shared reduce-mode member path — the same fn reproduction re-runs.
            run_blueprint_member(reload(doc), point, &space, data.run_sources(env, &binding.columns()), window, 0, pip, &topo, env, StopRule::Vol { length: R_SMA_STOP_LENGTH, k: R_SMA_STOP_K }, &binding, &[], NO_INSTRUMENT_CONTEXT)
        })
        // render the sweep terminal's BindError to prose (#247), the fn's String error
        // contract — never the raw Debug struct.
        .map_err(|e| render_bind_error(&e))
}

/// Sweep the LOADED blueprint over the user `--axis` grid on an in-sample window
/// `[from,to]` — the windowed, lattice-carrying twin of
/// `blueprint_sweep_family`. `sweep_with_lattice` gives the grid lattice `--select
/// plateau` needs. An unknown/kind-mismatched axis surfaces as `BindError` at the
/// sweep terminal (no panic, no hidden exit) for the caller to render. An axis
/// naming a bound param re-opens it (#246: bound value = default, same
/// `override_paths`/`reopen_all` recipe as `blueprint_sweep_family` — this is
/// its walk-forward in-sample twin); an axis matching neither space is refused
/// (wrapped as `BindError::UnknownKnob`, the honest replacement for the retired
/// "fully bound; nothing to sweep" refusal) before any member runs.
fn blueprint_sweep_over(
    doc: &str, axes: &[(String, Vec<Scalar>)], from: Timestamp, to: Timestamp, data: &DataSource,
    env: &project::Env, binding: &binding::ResolvedBinding,
) -> Result<(SweepFamily, Vec<usize>), BindError> {
    let reload = |d: &str| {
        blueprint_from_json(d, &|t| env.resolve(t))
            .expect("doc parse-validated at the dispatch boundary; reload is infallible")
    };
    let pip = data.pip_size();
    let probe_signal = reload(doc);
    let topo = topology_hash(&probe_signal);
    // The un-reopened wrapped OPEN space (#246), against a raw probe + raw strategy
    // load: derives the override set (which named axes re-open a bound param) before
    // the reopened probe is built — probe and per-member reloads must re-open
    // identically so points resolve against one space, exactly like
    // `blueprint_sweep_family`.
    let raw_space = blueprint_axis_probe(doc, env).param_space();
    let overrides = override_paths(axes, &raw_space, &probe_signal).map_err(BindError::UnknownKnob)?;
    let probe = blueprint_axis_probe_reopened(doc, env, &overrides);
    let space = probe.param_space();
    let mut iter = axes.iter();
    let (first_name, first_vals) = iter.next().expect("a blueprint walk-forward declares >= 1 axis");
    let mut binder = probe.axis(first_name, first_vals.clone());
    for (n, vals) in iter {
        binder = binder.axis(n, vals.clone());
    }
    binder.sweep_with_lattice(|point| {
        let sources = data.windowed_sources(from, to, env, &binding.columns());
        let window = window_of(&sources).expect("non-empty in-sample window");
        run_blueprint_member(reopen_all(reload(doc), &overrides), point, &space, sources, window, 0, pip, &topo, env, StopRule::Vol { length: R_SMA_STOP_LENGTH, k: R_SMA_STOP_K }, binding, &[], NO_INSTRUMENT_CONTEXT)
    })
}

/// Run the winner params over an out-of-sample window `[from,to]` on the loaded
/// blueprint. The reduce-mode member
/// (`run_blueprint_member`) retains R-metrics, not a raw pip curve, so the stitching
/// segment is empty (an empty segment leaves the stitched curve unbroken). `overrides`
/// (#246) is the SAME family-wide set `blueprint_walkforward_family` derived once and
/// resolved `space`/`params` against — the OOS reload must re-open it too, or a
/// bound-param axis's winner point (kind-checked against the REOPENED space) fails
/// `bootstrap_with_cells`'s arity check against this still-closed reload.
#[allow(clippy::too_many_arguments)]
fn run_oos_blueprint(
    doc: &str, params: &[Cell], space: &[ParamSpec], from: Timestamp, to: Timestamp,
    topo: &str, data: &DataSource, env: &project::Env, binding: &binding::ResolvedBinding,
    overrides: &[String],
) -> (Vec<(Timestamp, f64)>, RunReport) {
    let reload = reopen_all(
        blueprint_from_json(doc, &|t| env.resolve(t))
            .expect("doc parse-validated at the dispatch boundary; reload is infallible"),
        overrides,
    );
    let pip = data.pip_size();
    let sources = data.windowed_sources(from, to, env, &binding.columns());
    let window = window_of(&sources).expect("non-empty out-of-sample window");
    let report = run_blueprint_member(reload, params, space, sources, window, 0, pip, topo, env, StopRule::Vol { length: R_SMA_STOP_LENGTH, k: R_SMA_STOP_K }, binding, &[], NO_INSTRUMENT_CONTEXT);
    (Vec::new(), report)
}

/// A constant, zero-compute `RunReport` for [`validate_axis_grid`]'s sweep-terminal
/// probe. `SweepBinder::sweep_with_lattice`'s own `resolve_axes`/arity/kind checks all
/// run BEFORE this closure is invoked per grid point, so its body never influences the
/// validation outcome — only its signature (`Fn(&[Cell]) -> RunReport`) needs to
/// satisfy the terminal, at O(1) cost per point instead of a full member run.
fn axis_grid_probe_report() -> RunReport {
    RunReport {
        manifest: RunManifest {
            commit: String::new(),
            params: Vec::new(),
            defaults: Vec::new(),
            window: (Timestamp(0), Timestamp(0)),
            seed: 0,
            broker: "wf-axis-preflight-placeholder".to_string(),
            selection: None,
            instrument: None,
            topology_hash: None,
            project: None,
        },
        metrics: RunMetrics { total_pips: 0.0, max_drawdown: 0.0, bias_sign_flips: 0, r: None },
    }
}

/// Validate the `--axis` grid against `doc`'s wrapped param space WITHOUT running any
/// member (#253). Reuses the SAME strict, erroring axis-name check
/// `blueprint_sweep_over` performs (`override_paths` — single-sourced, so the two
/// paths cannot drift to differently-worded rejections) to derive the #246 override
/// set, then drives the sweep terminal's own `resolve_axes`/arity/kind checks
/// (`SweepBinder::sweep_with_lattice`) with [`axis_grid_probe_report`] standing in for
/// the run closure — no data access, no sim engine tick. Axis resolution is
/// window-agnostic, so the caller derives or passes no window at all.
fn validate_axis_grid(
    doc: &str, axes: &[(String, Vec<Scalar>)], raw_space: &[ParamSpec], probe_signal: &Composite,
    env: &project::Env,
) -> Result<(), BindError> {
    let overrides = override_paths(axes, raw_space, probe_signal).map_err(BindError::UnknownKnob)?;
    let probe = blueprint_axis_probe_reopened(doc, env, &overrides);
    let mut iter = axes.iter();
    let (first_name, first_vals) = iter.next().expect("a blueprint walk-forward declares >= 1 axis");
    let mut binder = probe.axis(first_name, first_vals.clone());
    for (n, vals) in iter {
        binder = binder.axis(n, vals.clone());
    }
    binder.sweep_with_lattice(|_| axis_grid_probe_report()).map(|_| ())
}

/// The loaded-blueprint IS-refit walk-forward: per IS window, re-optimize the
/// blueprint over the user `--axis` grid, select by `sqn_normalized`, run the
/// winner OOS, reusing the generic `walk_forward` driver + `select_winner`;
/// only the per-window sweep/OOS source the loaded blueprint. In-closure errors
/// (a bad `--axis`) `exit(2)` with the sweep terminal's message.
fn blueprint_walkforward_family(
    doc: &str, axes: &[(String, Vec<Scalar>)], data: &DataSource, select: Selection,
    env: &project::Env,
) -> WalkForwardResult {
    let span = data.wf_full_span(env);
    let (is_len, oos_len, step) = data.wf_window_sizes();
    let roller = match WindowRoller::new(span, is_len, oos_len, step, RollMode::Rolling) {
        Ok(r) => r,
        Err(e) => {
            eprintln!("aura: walk-forward window too short for one IS+OOS span: {e:?}");
            std::process::exit(2);
        }
    };
    let probe_signal = blueprint_from_json(doc, &|t| env.resolve(t))
        .expect("doc parse-validated at the dispatch boundary; reload is infallible");
    let topo = topology_hash(&probe_signal);
    // The un-reopened wrapped OPEN space (#246), against a raw probe + raw strategy
    // load: derives the override set ONCE for the whole family — every per-window
    // sweep AND the OOS reload re-open the SAME set, mirroring
    // `blueprint_sweep_family`/`blueprint_sweep_over`. The SILENT variant
    // (`wrapped_bound_overrides_of`, not the validating `override_paths`): an axis
    // matching neither space is simply not an override here — the strict check + its
    // established error message stay single-sourced in `blueprint_sweep_over`'s own
    // pre-flight call below, so this derivation cannot double-validate with a
    // differently-worded rejection.
    let axis_names: Vec<String> = axes.iter().map(|(n, _)| n.clone()).collect();
    let raw_space = blueprint_axis_probe(doc, env).param_space();
    let overrides = wrapped_bound_overrides_of(&axis_names, &raw_space, &probe_signal);
    let space = blueprint_axis_probe_reopened(doc, env, &overrides).param_space();
    // Strict binding resolution, once per family; refusal is the established
    // `aura: ` + exit-1 register (the roller's usage refusals stay exit 2).
    let binding = binding::resolve_binding(probe_signal.name(), probe_signal.input_roles(), &BTreeMap::new())
        .unwrap_or_else(|m| {
            eprintln!("aura: {m}");
            std::process::exit(1);
        });
    if matches!(data, DataSource::Synthetic) && !binding.close_only() {
        eprintln!("aura: {}", binding::synthetic_refusal(probe_signal.name(), &binding));
        std::process::exit(1);
    }
    // Validate the `--axis` grid ONCE at the dispatch boundary, mirroring `aura sweep`
    // (which resolves its axes a single time before any member runs). `walk_forward` fans
    // the per-window closure out across the windows in parallel, so a `BindError` raised
    // *inside* the closure would `eprintln!`+`exit(2)` from several windows before any one
    // exit lands — a racy, duplicated rejection (#177). Axis resolution is window-agnostic
    // (#253): `validate_axis_grid` resolves the SAME grid the sweep terminal would, without
    // running a single member — no IS window (or a second roller) is needed here at all.
    if let Err(e) = validate_axis_grid(doc, axes, &raw_space, &probe_signal, env) {
        eprintln!("aura: {}", render_bind_error(&e));
        std::process::exit(2);
    }
    walk_forward(roller, space.clone(), |w: WindowBounds| {
        let (is_family, lattice) = blueprint_sweep_over(doc, axes, w.is.0, w.is.1, data, env, &binding)
            .expect("axes validated in the dispatch-boundary pre-flight");
        let (best, selection) = match select_winner(&is_family, WINNER_SELECTION_METRIC, select, Some(&lattice)) {
            Ok(v) => v,
            Err(msg) => { eprintln!("aura: {msg}"); std::process::exit(2); }
        };
        let (oos_equity, mut oos_report) =
            run_oos_blueprint(doc, &best.params, &space, w.oos.0, w.oos.1, &topo, data, env, &binding, &overrides);
        oos_report.manifest.selection = Some(selection);
        WindowRun { chosen_params: best.params, oos_equity, oos_report }
    })
}

/// A fresh seeded synthetic price walk for one Monte-Carlo draw — `blueprint_mc_family`'s
/// pattern (a distinct realization per seed). A FIXED `SyntheticSpec` shared by the
/// `aura mc <blueprint.json>` persist path AND the reproduce MonteCarlo branch, so the
/// seed->walk reconstruction is bit-exact (C1). Length 60 comfortably warms the loaded
/// r-sma graph (SMA slow=4 + the len-3 vol stop) so draws produce differing trades.
fn synthetic_walk_sources(seed: u64) -> Vec<Box<dyn aura_engine::Source>> {
    let spec = SyntheticSpec { start: 1.0, len: 60, step: 1 };
    vec![Box::new(spec.source(seed))]
}

/// Build a Monte-Carlo family from a loaded CLOSED signal blueprint: run the fixed
/// blueprint across `n_seeds` seeds, each seed drawing a distinct synthetic walk. The
/// blueprint must be CLOSED (empty wrapped `param_space`) — MC binds no axis, so a free
/// knob has no binder; an OPEN blueprint yields a named `Err` (exit-free like the sibling
/// [`blueprint_sweep_family`]: the IO wrapper [`run_blueprint_mc`] renders it to stderr +
/// exit 2) before any run, pre-empting the `compile_with_params` arity panic. Each draw
/// runs the shared reduce-mode member path (`run_blueprint_member`, the same fn reproduce
/// re-runs), so reproduction is bit-identical (C1); every member carries the shared
/// `topology_hash`.
fn blueprint_mc_family(
    doc: &str, n_seeds: u64, data: &DataSource, env: &project::Env,
) -> Result<McFamily, String> {
    let reload = |d: &str| {
        blueprint_from_json(d, &|t| env.resolve(t))
            .expect("doc parse-validated at the dispatch boundary; reload is infallible")
    };
    let probe_signal = reload(doc);
    let topo = topology_hash(&probe_signal);
    // Strict binding resolution (name defaults — mc's synthetic family binds
    // no campaign overrides); the exit-free Err contract of this builder.
    let binding = binding::resolve_binding(probe_signal.name(), probe_signal.input_roles(), &BTreeMap::new())?;
    if !binding.close_only() {
        // MC draws ALWAYS run the seeded synthetic close walk (real-data mc
        // routes through the campaign sugar and never reaches this builder).
        return Err(binding::synthetic_refusal(probe_signal.name(), &binding));
    }
    let pip = data.pip_size();
    // probe the wrapped param_space (the same probe the sweep resolves against);
    // MC needs it empty. `blueprint_axis_probe` is the single source of that wrap.
    let space = blueprint_axis_probe(doc, env).param_space();
    if !space.is_empty() {
        // Exit-free like blueprint_sweep_family: the builder's single error contract is this
        // returned message (no hidden process exit), so the rejection is unit-testable; the IO
        // wrapper run_blueprint_mc renders it to stderr + exit 2 at the boundary.
        return Err(format!(
            "mc requires a closed blueprint (no free parameters); {} free knob(s) — \
             bind them or use `aura sweep --axis`",
            space.len()
        ));
    }
    // Closed blueprint -> an empty base point (as `aura run <blueprint.json>`); the MC
    // draws vary the SEED, not a tuning param (C12 axis 4). Delegate the disjoint C1 draws
    // to the shared `monte_carlo` helper — it runs them in parallel across sims (invariant 1),
    // deterministic in seed-input order. Each draw
    // re-runs the shared reduce-mode member path over its own seeded synthetic walk.
    let seeds: Vec<u64> = (1..=n_seeds).collect();
    let base_point: Vec<Scalar> = Vec::new();
    let family = monte_carlo(&base_point, &seeds, |seed, _base| {
        let sources = synthetic_walk_sources(seed);
        let window = window_of(&sources).expect("non-empty synthetic walk");
        run_blueprint_member(reload(doc), &[], &space, sources, window, seed, pip, &topo, env, StopRule::Vol { length: R_SMA_STOP_LENGTH, k: R_SMA_STOP_K }, &binding, &[], NO_INSTRUMENT_CONTEXT)
    });
    // Silent-vacuous MC guard (refuse-don't-guess, C10): with >= 2 seeds, if every draw's
    // metrics are bit-identical to the first, no seed reached a distinguishable realization —
    // the strategy never warmed over the fixed synthetic walk (e.g. a lookback as deep as the
    // walk is long), so the "distribution" is a single point masquerading as a family: a wrong
    // result with no error. Compare `metrics`, not the whole `RunReport` — the manifest's
    // `seed` differs per draw by construction, so a whole-report compare could never detect the
    // collapse; the metrics are the realization the seed is meant to move. A single-draw MC
    // (n == 1) is trivially "all identical" and is NOT this cross-seed condition, so it passes.
    if family.draws.len() >= 2
        && family
            .draws
            .iter()
            .all(|d| d.report.metrics == family.draws[0].report.metrics)
    {
        return Err(
            "mc is vacuous: every seed produced an identical result — the strategy never warmed \
             over the synthetic walk, so no seed reached a distinguishable realization; use a \
             shallower-lookback blueprint or a longer walk"
                .to_string(),
        );
    }
    Ok(family)
}

/// `aura sweep <blueprint.json> --axis <name>=<csv> …`: sweep a loaded signal over its
/// named param-space axes (the cycle-2 World/C21 verb). Builds the family via
/// [`blueprint_sweep_family`], surfaces an unknown / kind-mismatched axis as a named
/// error (stderr + exit 2, never a panic), ALWAYS records it as a `FamilyKind::Sweep`
/// family (C18/C21 lineage, exactly as the other family verbs), and prints each member
/// carrying the assigned `family_id` via [`family_member_line`] — so a printed member is
/// linkable back to its stored family, like `run_blueprint_walkforward` / `run_blueprint_mc`.
///
/// Divergence from the retired `run_sweep` (the one place this does less): the blueprint sweep is
/// reduce-only this cycle — [`blueprint_sweep_family`] writes no per-member traces — so
/// `persist`/`--trace` neither writes trace files nor reserves a trace-store name (that
/// reservation would guard a write that never happens, and could spuriously reject a
/// valid sweep on a name collision). `persist` is therefore not yet load-bearing here;
/// it is retained for the deferred per-member trace path. The family record itself is
/// written unconditionally, so lineage (C18/C21) holds whether or not `--trace` is given.
///
/// Synthetic-only: real-data invocations route through
/// `verb_sugar::run_sweep_sugar` at the dispatch boundary and never reach
/// this fn.
fn run_blueprint_sweep(
    doc: &str, axes: &[(String, Vec<Scalar>)], name: &str, persist: bool, data: DataSource,
    env: &project::Env,
) {
    let _ = persist; // reserved for the deferred per-member trace path; the family record below is unconditional
    let family = blueprint_sweep_family(doc, axes, &data, env).unwrap_or_else(|e| {
        eprintln!("aura: {e}");
        std::process::exit(2);
    });
    let reg = env.registry();
    // Store the canonical blueprint ONCE, keyed by the family's shared topology_hash —
    // exactly the bytes whose SHA256 the members carry (#164 byte-canonical, round-trip
    // idempotent). One stored topology per family (C18/C11/C12).
    let topo = family.points[0]
        .report
        .manifest
        .topology_hash
        .clone()
        .expect("a blueprint sweep stamps every member's topology_hash");
    let canonical = blueprint_to_json(
        &blueprint_from_json(doc, &|t| env.resolve(t))
            .expect("doc parse-validated at the dispatch boundary"),
    )
    .expect("a loaded blueprint re-serializes");
    reg.put_blueprint(&topo, &canonical).unwrap_or_else(|e| {
        eprintln!("aura: {e}");
        std::process::exit(1);
    });
    // Record the family unconditionally (C18/C21 lineage), exactly like
    // `run_blueprint_walkforward` / `run_blueprint_mc`.
    let id = match reg.append_family(name, FamilyKind::Sweep, &sweep_member_reports(&family)) {
        Ok(id) => id,
        Err(e) => {
            eprintln!("aura: {e}");
            std::process::exit(1);
        }
    };
    for pt in &family.points {
        println!("{}", family_member_line(&id, &pt.report));
    }
}

/// `aura walkforward <blueprint.json> --axis …`: build the loaded-blueprint
/// IS-refit walk-forward, store the canonical blueprint ONCE keyed by the shared
/// `topology_hash` (the C18 hook, so `aura reproduce` re-derives it), record it as
/// a `FamilyKind::WalkForward` family, and print each OOS member line + the summary.
/// Mirrors `run_blueprint_sweep` (content-addressed family verb; no ensure_name_free).
fn run_blueprint_walkforward(
    doc: &str, axes: &[(String, Vec<Scalar>)], name: &str, data: DataSource, select: Selection,
    env: &project::Env,
) {
    let result = blueprint_walkforward_family(doc, axes, &data, select, env);
    let reg = env.registry();
    let topo = result.windows[0]
        .run
        .oos_report
        .manifest
        .topology_hash
        .clone()
        .expect("a blueprint walk-forward stamps every member's topology_hash");
    let canonical = blueprint_to_json(
        &blueprint_from_json(doc, &|t| env.resolve(t)).expect("doc parse-validated at the dispatch boundary"),
    )
    .expect("a loaded blueprint re-serializes");
    reg.put_blueprint(&topo, &canonical).unwrap_or_else(|e| { eprintln!("aura: {e}"); std::process::exit(1); });
    let id = match reg.append_family(name, FamilyKind::WalkForward, &walkforward_member_reports(&result)) {
        Ok(id) => id,
        Err(e) => { eprintln!("aura: {e}"); std::process::exit(1); }
    };
    for w in &result.windows {
        println!("{}", family_member_line(&id, &w.run.oos_report));
    }
    println!("{}", walkforward_summary_json(&result));
}

/// `aura mc <blueprint.json> --seeds N`: build a Monte-Carlo family from a loaded CLOSED
/// blueprint (the World/C21 verb), store the canonical blueprint ONCE keyed by the shared
/// `topology_hash` (the 0094 hook, so `aura reproduce` re-derives it), record it as a
/// `FamilyKind::MonteCarlo` family (C18/C21 lineage), and print each draw's member line
/// (carrying the seed) plus the aggregate — mirroring `run_blueprint_sweep`.
fn run_blueprint_mc(doc: &str, n_seeds: u64, name: &str, data: DataSource, env: &project::Env) {
    let family = blueprint_mc_family(doc, n_seeds, &data, env).unwrap_or_else(|e| {
        eprintln!("aura: {e}");
        std::process::exit(2);
    });
    let reg = env.registry();
    // Store the canonical blueprint ONCE, keyed by the family's shared topology_hash.
    let topo = family.draws[0]
        .report
        .manifest
        .topology_hash
        .clone()
        .expect("a blueprint mc stamps every member's topology_hash");
    let canonical = blueprint_to_json(
        &blueprint_from_json(doc, &|t| env.resolve(t))
            .expect("doc parse-validated at the dispatch boundary"),
    )
    .expect("a loaded blueprint re-serializes");
    reg.put_blueprint(&topo, &canonical).unwrap_or_else(|e| {
        eprintln!("aura: {e}");
        std::process::exit(1);
    });
    let id = match reg.append_family(name, FamilyKind::MonteCarlo, &mc_member_reports(&family)) {
        Ok(id) => id,
        Err(e) => {
            eprintln!("aura: {e}");
            std::process::exit(1);
        }
    };
    for draw in &family.draws {
        println!("{}", mc_member_line(&id, draw.seed, &draw.report));
    }
    println!("{}", mc_aggregate_json(&family.aggregate));
}

/// The Donchian channel length for the canonical closed r_breakout example. Single
/// source for the emitter and the three proof tests that must all agree with the
/// baked `examples/r_breakout.json` — kept honest by one constant instead of a
/// hand-synced `Some(3)` literal recurring across those call sites.
#[cfg(test)]
const R_BREAKOUT_CHANNEL: i64 = 3;

/// The Donchian breakout signal leg, a pure `price→bias` Composite so it serialises
/// as blueprint data (#159 cut 2). The signal computation matches the retired fused
/// builder's leg (same nodes, same wiring); the pip/R harness is the generic
/// `wrap_r` wrapper, not part of the signal. `channel = Some(n)` binds both rolling
/// nodes (closed); `None` leaves them open, ganged into the single `channel_length`
/// knob (#61) — the channel is structurally ONE parameter. This carve has no production
/// (non-test) caller — it is exercised only by the regeneration + proof tests below;
/// production code loads the shipped JSON, not this builder — hence `#[cfg(test)]`.
#[cfg(test)]
fn r_breakout_signal(channel: Option<i64>) -> Composite {
    let mut g = GraphBuilder::new("r_breakout_signal");
    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);
    if channel.is_none() {
        g.gang("channel_length", [mx.param("length"), mn.param("length")]);
    }
    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}
    let price = g.source_role("price", 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"));
    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.expose(exposure.output("value"), "bias");
    g.build().expect("r_breakout signal wiring resolves")
}

/// The EWMA window for the canonical closed r_meanrev example (ganged mean/var Ema
/// length; the open form gangs them structurally via the single `window` knob, #61).
/// Single source for the emitter and the proof tests that must all agree with the
/// baked `examples/r_meanrev.json`.
#[cfg(test)]
const R_MEANREV_WINDOW: i64 = 3;

/// The Bollinger band half-width (in sigma) for the canonical closed r_meanrev
/// example. Single source alongside [`R_MEANREV_WINDOW`].
#[cfg(test)]
const R_MEANREV_BAND_K: f64 = 2.0;

/// The EWMA Bollinger-band mean-reversion signal leg, carved out of the retired fused
/// builder as a pure `price→bias` Composite so it serialises as blueprint data (#159
/// cut 3). Verbatim signal computation vs that retired fused builder, except the band
/// half-width `band_k*sigma` uses `Scale` (a rosterable multiply) in place of
/// `LinComb(1)`. `#[cfg(test)]`: its only role is regenerating + pinning the examples;
/// production loads the shipped JSON.
#[cfg(test)]
fn r_meanrev_signal(window: Option<i64>, band_k: Option<f64>) -> Composite {
    let mut g = GraphBuilder::new("r_meanrev_signal");
    let (mut mean_b, mut var_b) =
        (Ema::builder().named("mean_window"), Ema::builder().named("var_window"));
    if let Some(n) = window {
        mean_b = mean_b.bind("length", Scalar::i64(n));
        var_b = var_b.bind("length", Scalar::i64(n));
    }
    let mean = g.add(mean_b);
    let dev = g.add(Sub::builder()); // price - mean
    let sq = g.add(Mul::builder()); // dev * dev
    let var = g.add(var_b); // EWMA variance
    if window.is_none() {
        g.gang("window", [mean.param("length"), var.param("length")]);
    }
    let sigma = g.add(Sqrt::builder());
    let mut band_b = Scale::builder().named("band");
    if let Some(k) = band_k {
        band_b = band_b.bind("factor", Scalar::f64(k));
    }
    let band = g.add(band_b);
    let upper = g.add(Add::builder());
    let lower = g.add(Sub::builder());
    let gt_hi = g.add(Gt::builder());
    let gt_lo = g.add(Gt::builder());
    let short_latch = g.add(Latch::builder());
    let long_latch = g.add(Latch::builder());
    let exposure = g.add(Sub::builder()); // long_latch - short_latch -> bias
    let price = g.source_role("price", ScalarKind::F64);
    g.feed(price, [mean.input("series"), dev.input("lhs"), gt_hi.input("a"), gt_lo.input("b")]);
    g.connect(mean.output("value"), dev.input("rhs"));
    g.connect(dev.output("value"), sq.input("lhs"));
    g.connect(dev.output("value"), sq.input("rhs"));
    g.connect(sq.output("value"), var.input("series"));
    g.connect(var.output("value"), sigma.input("value"));
    g.connect(sigma.output("value"), band.input("signal"));
    g.connect(mean.output("value"), upper.input("lhs"));
    g.connect(band.output("value"), upper.input("rhs"));
    g.connect(mean.output("value"), lower.input("lhs"));
    g.connect(band.output("value"), lower.input("rhs"));
    g.connect(upper.output("value"), gt_hi.input("b"));
    g.connect(lower.output("value"), gt_lo.input("a"));
    g.connect(gt_hi.output("value"), short_latch.input("set"));
    g.connect(gt_lo.output("value"), short_latch.input("reset"));
    g.connect(gt_lo.output("value"), long_latch.input("set"));
    g.connect(gt_hi.output("value"), long_latch.input("reset"));
    g.connect(long_latch.output("value"), exposure.input("lhs"));
    g.connect(short_latch.output("value"), exposure.input("rhs"));
    g.expose(exposure.output("value"), "bias");
    g.build().expect("r_meanrev signal wiring resolves")
}

/// The channel length for the canonical closed r_channel example. Single
/// source for the emitter and the proof tests that must agree with the baked
/// `examples/r_channel.json` (mirrors [`R_BREAKOUT_CHANNEL`]).
#[cfg(test)]
const R_CHANNEL_LENGTH: i64 = 3;

/// The OHLC high/low-channel (Donchian-shape) signal leg — the harness-input-
/// binding acceptance strategy (#231): bias goes long when the CLOSE breaks
/// the rolling max of the previous `n` HIGHS, short when it breaks the rolling
/// min of the previous `n` LOWS. Three input roles (`high`, `low`, `close` —
/// the role names ARE the column binding), declared in canonical column order.
/// `channel = Some(n)` binds both rolling nodes (closed); `None` leaves them
/// open, ganged into the single `channel_length` knob (#61). `#[cfg(test)]`:
/// production loads the shipped JSON; this builder only regenerates + pins it
/// (the r_breakout/r_meanrev carve pattern).
#[cfg(test)]
fn r_channel_signal(channel: Option<i64>) -> Composite {
    let mut g = GraphBuilder::new("hl_channel");
    let delay_hi = g.add(Delay::builder().named("prev_high").bind("lag", Scalar::i64(1)));
    let delay_lo = g.add(Delay::builder().named("prev_low").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);
    if channel.is_none() {
        g.gang("channel_length", [mx.param("length"), mn.param("length")]);
    }
    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}
    let high = g.source_role("high", ScalarKind::F64);
    let low = g.source_role("low", ScalarKind::F64);
    let close = g.source_role("close", ScalarKind::F64);
    g.feed(high, [delay_hi.input("series")]);
    g.feed(low, [delay_lo.input("series")]);
    g.feed(close, [gt_up.input("a"), gt_down.input("b")]);
    g.connect(delay_hi.output("value"), mx.input("series"));
    g.connect(delay_lo.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.expose(exposure.output("value"), "bias");
    g.build().expect("hl_channel signal wiring resolves")
}

/// Parse the `--params` value: a JSON array of externally-tagged `Scalar` cells
/// (`[{"I64":2},{"F64":0.5}]`, the #155 wire form `Scalar` derives via serde). The cells
/// bind positionally against the loaded signal's `param_space`. A malformed array is
/// refused with the flag named, never silently coerced to an empty param vector
/// (refuse-don't-guess, C10) — a dropped param would bootstrap a *different* graph.
fn parse_param_cells(json: &str) -> Result<Vec<Scalar>, String> {
    serde_json::from_str(json).map_err(|e| format!("--params: {e}"))
}

/// Lex a `--axis` CSV into typed Scalars by shape: an integer-shaped token is i64,
/// otherwise f64. `resolve_axes` kind-checks each value against the param's declared
/// kind afterwards (a mismatch is a named error, not a panic).
fn parse_scalar_csv(csv: &str) -> Option<Vec<Scalar>> {
    csv.split(',').map(|t| {
        let t = t.trim();
        if t.is_empty() { return None; }
        match t.parse::<i64>() {
            Ok(i) => Some(Scalar::i64(i)),
            Err(_) => t.parse::<f64>().ok().map(Scalar::f64),
        }
    }).collect()
}

// ============================== clap parser surface ==============================
// The declarative argument grammar. clap owns argv tokenizing, scoped `--help`,
// `--version`, `--flag=value`, `--`, and long-option abbreviation; the `dispatch_*`
// handlers below convert each `*Cmd` into the argument shapes the existing execution
// fns accept, reusing the value helpers (`Selection`,
// `DataSource::from_choice`, `parse_scalar_csv`, `parse_csv_list`, `parse_select`,
// `parse_param_cells`). The four subcommands with an optional `[blueprint]` positional
// dispatch on `is_blueprint_file`: a first-positional naming an existing `.json` file
// selects the loaded-blueprint branch. There is no second built-in grammar anymore
// (#159 demo retirement / #220 axis generalization): without a blueprint the
// dispatcher prints a usage error and exits 2; a blueprint with `--real` routes to
// the campaign path. Usage errors exit 2; runtime refusals exit 1; a run that
// completes with ≥1 failed cell exits 3 (#272); a clean run exits 0.

/// The single build-time commit provenance (`option_env!("AURA_COMMIT")`,
/// falling back to `"unknown"`): both `--version` (via `version_string`) and
/// `sim_optimal_manifest`'s `RunManifest.commit` read this one const, so a
/// stale binary is distinguishable from a fresh one at the CLI surface,
/// before any run, without a second commit-sourcing mechanism (#266).
const ENGINE_COMMIT: &str = match option_env!("AURA_COMMIT") {
    Some(c) => c,
    None => "unknown",
};

/// `aura 0.1.0 (<commit>)` as a process-lifetime `&'static str` — computed once
/// (clap's `version` builder method wants `Into<Str>`, and `clap::builder::Str`
/// (clap 4.6) only implements `From<&'static str>`, not `From<String>`, so the
/// one owned `String` this formats is leaked for the process's lifetime, same
/// as any other CLI one-shot startup string).
fn version_string() -> &'static str {
    static VERSION: std::sync::OnceLock<String> = std::sync::OnceLock::new();
    VERSION.get_or_init(|| format!("{} ({ENGINE_COMMIT})", env!("CARGO_PKG_VERSION")))
}

/// The `aura` root parser. `version` is built from `CARGO_PKG_VERSION` (the
/// workspace `0.1.0`) plus the parenthesized `ENGINE_COMMIT`, so
/// `aura --version` prints `aura 0.1.0 (<commit>)`.
#[derive(Parser)]
#[command(
    name = "aura",
    version = version_string(),
    about = "Author, backtest, and validate trading strategies — research CLI",
    infer_long_args = true
)]
struct Cli {
    #[command(subcommand)]
    command: Command,
    /// Load the project dylib from target/release instead of target/debug.
    #[arg(long, global = true)]
    release: bool,
}

#[derive(Subcommand)]
enum Command {
    /// Run a single backtest (built-in harness or a loaded blueprint).
    Run(RunCmd),
    /// Render a recorded run's trace to an HTML chart.
    Chart(ChartCmd),
    /// Emit / construct / introspect a graph.
    Graph(GraphCmd),
    /// Sweep a parameter grid over a strategy or a loaded blueprint.
    Sweep(SweepCmd),
    /// Walk-forward validation over a strategy or a loaded blueprint. Over --real, the
    /// fixed 90/30-day roller fits to a --from/--to window shorter than it, preserving
    /// the 3:1 IS:OOS ratio, instead of refusing the window outright.
    Walkforward(WalkforwardCmd),
    /// Grade one candidate across multiple instruments.
    Generalize(GeneralizeCmd),
    /// Monte-Carlo over synthetic draws, an R-bootstrap, or a loaded blueprint. Over
    /// --real, the fixed 90/30-day walk-forward roller fits to a --from/--to window
    /// shorter than it, preserving the 3:1 IS:OOS ratio, instead of refusing outright.
    Mc(McCmd),
    /// List or inspect recorded run families.
    Runs(RunsCmd),
    /// Reproduce a recorded family by content id.
    Reproduce(ReproduceCmd),
    /// Scaffold a new data-only research project.
    New(NewCmd),
    /// Scaffold and attach node crates (native-node development).
    Nodes(NodesCmd),
    /// Validate, introspect, and register process documents (methodology).
    Process(research_docs::ProcessCmd),
    /// Validate, introspect, and register campaign documents (experiment intent).
    Campaign(research_docs::CampaignCmd),
    /// Inspect the project's data archive.
    Data(DataCmd),
}

#[derive(Args)]
struct ChartCmd {
    /// The recorded run or family name to chart.
    name: String,
    /// Chart only the given tap.
    #[arg(long)]
    tap: Option<String>,
    /// Render stacked panels instead of an overlay.
    #[arg(long)]
    panels: bool,
}

#[derive(Args)]
struct NewCmd {
    /// Directory name to create.
    name: String,
}

#[derive(Args)]
struct NodesCmd {
    #[command(subcommand)]
    command: NodesCommand,
}

#[derive(Subcommand)]
enum NodesCommand {
    /// Scaffold a node crate beside this project and attach it via `[nodes]`.
    New(NodesNewCmd),
}

#[derive(Args)]
struct NodesNewCmd {
    /// Crate name; also the default namespace (dashes become underscores).
    name: String,
    /// Engine checkout for the crate's aura-core path-dep.
    #[arg(long)]
    engine_path: Option<std::path::PathBuf>,
    /// Vocabulary namespace override.
    #[arg(long)]
    namespace: Option<String>,
}

#[derive(Args)]
struct DataCmd {
    #[command(subcommand)]
    command: DataCommand,
}

#[derive(Subcommand)]
enum DataCommand {
    /// Report a symbol's monthly file-index coverage (span + interior gaps).
    Coverage(DataCoverageCmd),
    /// List the archive's known symbols, sorted, one per line.
    List,
}

#[derive(Args)]
struct DataCoverageCmd {
    /// The symbol to inventory (e.g. `GER40`).
    symbol: String,
}

#[derive(Args)]
#[command(args_conflicts_with_subcommands = true)]
struct GraphCmd {
    /// A blueprint .json file to render (omit for the built-in sample).
    blueprint: Option<String>,
    #[command(subcommand)]
    sub: Option<GraphSub>,
}

#[derive(Subcommand)]
enum GraphSub {
    /// Construct a graph from a stdin op-list.
    Build,
    /// Introspect a graph.
    Introspect(GraphIntrospectCmd),
    /// Register a blueprint document into the content-addressed store (#196).
    Register {
        /// The blueprint .json file to register.
        file: std::path::PathBuf,
    },
}

#[derive(Args)]
struct GraphIntrospectCmd {
    /// List the closed node vocabulary (one node type per line).
    #[arg(long)]
    vocabulary: bool,
    /// Describe one node type's ports by name.
    #[arg(long)]
    node: Option<String>,
    /// List the graph's unwired (unbound) ports.
    #[arg(long)]
    unwired: bool,
    /// Print the graph's content id (topology hash). With FILE, read the
    /// document from it (a blueprint envelope or an op-list, shape-
    /// discriminated, #196); without FILE, read a stdin op-list as before.
    #[arg(long, value_name = "FILE")]
    content_id: Option<Option<std::path::PathBuf>>,
    /// Print the graph's topology-identity id (debug names stripped).
    #[arg(long)]
    identity_id: bool,
    /// Print a blueprint's raw param space (the campaign-axis namespace), one
    /// name:kind line per open param. FILE path or 64-hex store content id (#196).
    #[arg(long, value_name = "FILE|ID")]
    params: Option<String>,
}

#[derive(Args)]
struct GeneralizeCmd {
    /// The candidate blueprint (.json, required) — graded across instruments.
    blueprint: Option<String>,
    /// Comma-separated instrument list (>=2 distinct, required).
    #[arg(long)]
    real: Option<String>,
    /// Candidate axis `<name>=<value>` (repeatable, >=1; one value per axis).
    #[arg(long)]
    axis: Vec<String>,
    /// Candidate stop length (single value; optional, defaults to
    /// [`R_SMA_STOP_LENGTH`]).
    #[arg(long)]
    stop_length: Option<i64>,
    /// Candidate stop-k multiple (single value; optional, defaults to
    /// [`R_SMA_STOP_K`]).
    #[arg(long)]
    stop_k: Option<f64>,
    /// Window start (Unix ms, inclusive).
    #[arg(long)]
    from: Option<i64>,
    /// Window end (Unix ms, inclusive).
    #[arg(long)]
    to: Option<i64>,
    /// Grading metric (default expectancy_r).
    #[arg(long)]
    metric: Option<String>,
    /// Family name (default generalize).
    #[arg(long)]
    name: Option<String>,
}

#[derive(Args)]
struct RunsCmd {
    #[command(subcommand)]
    sub: RunsSub,
}

#[derive(Subcommand)]
enum RunsSub {
    /// List recorded families.
    Families,
    /// Inspect one family, optionally ranked by a metric.
    Family {
        id: String,
        /// The literal keyword `rank`, if a ranking metric follows.
        rank_kw: Option<String>,
        /// The metric to rank by (only valid after `rank`).
        metric: Option<String>,
    },
}

#[derive(Args)]
struct ReproduceCmd {
    /// The family content id to reproduce.
    id: String,
}

#[derive(Args)]
struct RunCmd {
    /// A serialized signal blueprint (.json). An existing file selects the
    /// loaded-blueprint grammar; otherwise the built-in harness grammar.
    blueprint: Option<String>,
    /// Blueprint params (JSON scalar-cell array; .json mode).
    #[arg(long)]
    params: Option<String>,
    /// Blueprint seed (.json mode).
    #[arg(long)]
    seed: Option<u64>,
    /// Real instrument symbol to backtest over (recorded data); omit for the synthetic stream.
    #[arg(long)]
    real: Option<String>,
    /// Window start (Unix ms, inclusive); requires --real.
    #[arg(long)]
    from: Option<i64>,
    /// Window end (Unix ms, inclusive); requires --real.
    #[arg(long)]
    to: Option<i64>,
    /// Not accepted — CLI-side trace persistence is retired (see #224).
    #[arg(long)]
    trace: Option<String>,
}

#[derive(Args)]
struct SweepCmd {
    /// A loaded blueprint (.json); omit for the built-in --strategy grammar.
    blueprint: Option<String>,
    /// Legacy `--strategy` selector: no built-in value remains (use a blueprint).
    /// Retired tokens fall to the generic usage error.
    #[arg(long)]
    strategy: Option<String>,
    /// Real instrument symbol to sweep over (recorded data); omit for the synthetic stream.
    #[arg(long)]
    real: Option<String>,
    /// Window start (Unix ms, inclusive); requires --real.
    #[arg(long)]
    from: Option<i64>,
    /// Window end (Unix ms, inclusive); requires --real.
    #[arg(long)]
    to: Option<i64>,
    /// Family name (records to the registry without persisting per-member traces).
    #[arg(long)]
    name: Option<String>,
    /// Family name that also persists each member's taps (--real mode; mutually
    /// exclusive with --name).
    #[arg(long)]
    trace: Option<String>,
    /// Blueprint sweep axis `<name>=<csv>` (repeatable; .json mode).
    #[arg(long)]
    axis: Vec<String>,
    /// List a loaded blueprint's sweepable axes and exit (.json mode, stands alone).
    #[arg(long)]
    list_axes: bool,
}

#[derive(Args)]
struct WalkforwardCmd {
    /// A loaded blueprint (.json, required — both grammars are blueprint-first).
    blueprint: Option<String>,
    /// Real instrument symbol to validate over (recorded data); omit for the synthetic stream.
    #[arg(long)]
    real: Option<String>,
    /// Window start (Unix ms, inclusive); requires --real.
    #[arg(long)]
    from: Option<i64>,
    /// Window end (Unix ms, inclusive); requires --real.
    #[arg(long)]
    to: Option<i64>,
    /// Family name (records to the registry without persisting per-member traces).
    #[arg(long)]
    name: Option<String>,
    /// Family name that also persists each OOS window's taps (--real mode;
    /// mutually exclusive with --name).
    #[arg(long)]
    trace: Option<String>,
    /// Campaign-path stop length (single value; --real mode).
    #[arg(long)]
    stop_length: Option<String>,
    /// Campaign-path stop-k multiple (single value; --real mode).
    #[arg(long)]
    stop_k: Option<String>,
    /// In-sample winner selection: argmax | plateau:mean | plateau:worst (default argmax).
    #[arg(long)]
    select: Option<String>,
    /// Blueprint IS-refit axis `<name>=<csv>` (repeatable, >=1 required; .json mode).
    #[arg(long)]
    axis: Vec<String>,
}

#[derive(Args)]
struct McCmd {
    /// A loaded blueprint (.json); omit for the built-in grammar.
    blueprint: Option<String>,
    /// Real instrument symbol for the R-bootstrap campaign path (recorded data); omit
    /// for the synthetic seed family.
    #[arg(long)]
    real: Option<String>,
    /// Window start (Unix ms, inclusive); requires --real.
    #[arg(long)]
    from: Option<i64>,
    /// Window end (Unix ms, inclusive); requires --real.
    #[arg(long)]
    to: Option<i64>,
    /// Family name for the synthetic seed-resweep (records without persisting traces).
    #[arg(long)]
    name: Option<String>,
    /// Not accepted with --real — the R-bootstrap records without a family name.
    #[arg(long)]
    trace: Option<String>,
    /// Blueprint IS-refit axis `<name>=<csv>` (repeatable, >=1 required; --real mode).
    #[arg(long)]
    axis: Vec<String>,
    /// Campaign-path stop length (single value; --real mode).
    #[arg(long)]
    stop_length: Option<String>,
    /// Campaign-path stop-k multiple (single value; --real mode).
    #[arg(long)]
    stop_k: Option<String>,
    /// Moving-block bootstrap block length (R-bootstrap; default 1).
    #[arg(long)]
    block_len: Option<usize>,
    /// Number of bootstrap resamples (R-bootstrap; default 1000).
    #[arg(long)]
    resamples: Option<usize>,
    /// Bootstrap RNG seed (R-bootstrap; default 1).
    #[arg(long)]
    seed: Option<u64>,
    /// Number of synthetic draws (required; .json mode).
    #[arg(long)]
    seeds: Option<u64>,
}

/// The dual-grammar discriminator: a first-positional that names an existing
/// `.json` file selects the loaded-blueprint branch. Single-sourced so the
/// four dual-grammar subcommands stay in lockstep.
fn is_blueprint_file(arg: &Option<String>) -> Option<&str> {
    arg.as_deref()
        .filter(|a| a.ends_with(".json") && std::path::Path::new(a).is_file())
}

/// Resolve a `[blueprint].json`-branch `--real`/`--from`/`--to` into a `RunData`,
/// mirroring the old `parse_blueprint_run_args` window guard (`--from`/`--to`
/// require `--real`; empty symbol rejected). Refuses in place (stderr + exit 2).
fn run_data_from(real: Option<&str>, from: Option<i64>, to: Option<i64>) -> RunData {
    let usage = "Usage: aura run <blueprint.json> [--params <json-cell-array>] [--seed <n>] [--real <SYMBOL> [--from <ms>] [--to <ms>]]";
    match real {
        Some(s) if !s.is_empty() => RunData::Real { symbol: s.to_string(), from, to },
        Some(_) => {
            eprintln!("aura: {usage}");
            std::process::exit(2);
        }
        None if from.is_some() || to.is_some() => {
            eprintln!("aura: {usage}");
            std::process::exit(2);
        }
        None => RunData::Synthetic,
    }
}

/// The real walk-forward roller sizes in ms (the campaign wf stage works in ms:
/// span `cell.window_ms`, sizes the `StageBlock` `_ms` fields). `WF_REAL_*_NS` are
/// nanoseconds; divide by 1e6. The ms sizes over the doc window reproduce the same
/// calendar windows the inline ns roller produces (anchor-gated).
fn wf_ms_sizes() -> (u64, u64, u64) {
    (
        (WF_REAL_IS_NS / 1_000_000) as u64,
        (WF_REAL_OOS_NS / 1_000_000) as u64,
        (WF_REAL_STEP_NS / 1_000_000) as u64,
    )
}

/// Fit the fixed real-archive walk-forward roller (`wf_ms_sizes`, 90/30/30 days)
/// to a resolved campaign window `(from_ms, to_ms)` (#239). When the fixed
/// IS+OOS span fits inside the window, the sizes come back byte-identical (every
/// existing anchor/e2e over a year-plus window pins this branch). When the window
/// is shorter than IS+OOS, the roller scales DOWN to the window, preserving the
/// fixed 3:1 IS:OOS ratio, with `step_ms == oos_ms` (a window of exactly IS+OOS
/// then yields exactly one roll — never zero). Pure and unit-testable; both
/// `dispatch_walkforward`/`dispatch_mc` consume it in place of the unconditional
/// `wf_ms_sizes()` stamp. The executor's own fit check (`campaign_run.rs`) stays
/// untouched — it still validates an AUTHORED document's declared window as-is.
fn fit_wf_ms_sizes(from_ms: i64, to_ms: i64) -> (u64, u64, u64) {
    let (is_ms, oos_ms, step_ms) = wf_ms_sizes();
    let span_ms = to_ms.saturating_sub(from_ms).max(0) as u64;
    if is_ms + oos_ms <= span_ms {
        return (is_ms, oos_ms, step_ms);
    }
    // Preserve the fixed IS:OOS ratio derived from the constants themselves (not a
    // bare literal): OOS = span/(ratio+1) (floor), IS = ratio*OOS, so
    // IS+OOS = (ratio+1)*OOS <= span always holds, and step == OOS (one roll over
    // the fit window, matching the fixed-size branch's `step_ms == oos_ms`).
    let ratio = is_ms / oos_ms;
    let oos_fit = span_ms / (ratio + 1);
    let is_fit = oos_fit * ratio;
    (is_fit, oos_fit, oos_fit)
}

/// Resolve a single optional stop knob (`--stop-length`/`--stop-k`): an absent flag
/// defaults to `default`; a present one parses via [`parse_csv_list`] and refuses
/// unless it holds exactly one value (the stop is a risk regime, not a swept axis).
/// Single-sourced across `walkforward_args_from`/`mc_args_from` (#217 follow-up) —
/// `generalize_args_from`'s typed `Option<i64>`/`Option<f64>` fields need no parse
/// step, so it keeps its own plain `unwrap_or`.
fn stop_knob_or<T: Copy + std::str::FromStr>(
    raw: Option<&str>,
    default: T,
    regime: &impl Fn() -> String,
) -> Result<T, String> {
    match raw {
        None => Ok(default),
        Some(s) => {
            let v: Vec<T> = parse_csv_list(s).map_err(|()| regime())?;
            if v.len() != 1 {
                return Err(regime());
            }
            Ok(v[0])
        }
    }
}

/// Convert `WalkforwardCmd` into the resolved argument shape the walkforward sugar
/// consumes (the dissolved `.json --real` branch). Single instrument, single-value
/// stop (Fork A: the stop is a risk regime, not a swept axis); `parse_csv_list` is
/// generic over `T: FromStr`, so the same helper parses the i64 length and the f64
/// k. `--stop-length`/`--stop-k` are each OPTIONAL (#217): a missing flag defaults
/// independently to the single-sourced [`R_SMA_STOP_LENGTH`]/[`R_SMA_STOP_K`]
/// regime; a present-but-multi-value stop still refuses (the local `regime`
/// closure, captureless → `Copy`, is reused across both arms). `--name`/`--trace`
/// are mutually exclusive, matching the flag's own documented contract and the
/// inline path's `name_persist` refusal; an omitted flag defaults the family name
/// to "walkforward". The IS-refit `--axis` grid is parsed separately at the
/// dispatch site, not here.
#[allow(clippy::type_complexity)]
fn walkforward_args_from(
    a: &WalkforwardCmd,
) -> Result<(String, bool, String, i64, f64, Option<i64>, Option<i64>), String> {
    let symbol = match a.real.as_deref() {
        None | Some("") => return Err("walkforward dissolves only over --real <SYMBOL>".to_string()),
        Some(s) => s.to_string(),
    };
    let regime = || "walkforward: the stop is a single risk regime; --stop-length and --stop-k take one value each".to_string();
    let stop_length = stop_knob_or(a.stop_length.as_deref(), R_SMA_STOP_LENGTH, &regime)?;
    let stop_k = stop_knob_or(a.stop_k.as_deref(), R_SMA_STOP_K, &regime)?;
    let (name, trace) = match (a.name.as_deref(), a.trace.as_deref()) {
        (Some(_), Some(_)) => {
            return Err("walkforward: --name and --trace are mutually exclusive".to_string());
        }
        (Some(n), None) => (n.to_string(), false),
        (None, Some(t)) => (t.to_string(), true),
        (None, None) => ("walkforward".to_string(), false),
    };
    Ok((name, trace, symbol, stop_length, stop_k, a.from, a.to))
}

/// Convert `McCmd` into the resolved argument shape the mc sugar consumes (the
/// `--real` campaign branch). Single instrument, single-value stop (Fork A: the
/// stop is a risk regime, not a swept axis); `--stop-length`/`--stop-k` are each
/// OPTIONAL (#217), independently defaulting to the single-sourced
/// [`R_SMA_STOP_LENGTH`]/[`R_SMA_STOP_K`] regime when omitted — a present-but-
/// multi-value stop still refuses. `--block-len`/`--resamples`/`--seed`
/// default to `1`/`1000`/`1` — the same defaults the retired real-R dispatch
/// used, so an omitted flag produces the same document either way. The
/// usize->u32 conversion lands HERE, at the argv boundary where the CLI's
/// `usize` meets the document's `u32` vocabulary (`StageBlock::MonteCarlo`).
/// `--name`/`--trace` are rejected: the R-bootstrap records without a family
/// name (the campaign name is a constant "mc").
#[allow(clippy::type_complexity)]
fn mc_args_from(
    a: &McCmd,
) -> Result<(String, String, i64, f64, u32, u32, u64, Option<i64>, Option<i64>), String> {
    let symbol = match a.real.as_deref() {
        None | Some("") => return Err("mc dissolves only over --real <SYMBOL>".to_string()),
        Some(s) => s.to_string(),
    };
    if a.name.is_some() || a.trace.is_some() {
        return Err("mc --real: --name/--trace are not accepted (the R-bootstrap records without a family name)".to_string());
    }
    let regime = || "mc: the stop is a single risk regime; --stop-length and --stop-k take one value each".to_string();
    let stop_length = stop_knob_or(a.stop_length.as_deref(), R_SMA_STOP_LENGTH, &regime)?;
    let stop_k = stop_knob_or(a.stop_k.as_deref(), R_SMA_STOP_K, &regime)?;
    let block_len = a.block_len.map(|v| v as u32).unwrap_or(1);
    let resamples = a.resamples.map(|v| v as u32).unwrap_or(1000);
    let seed = a.seed.unwrap_or(1);
    Ok(("mc".to_string(), symbol, stop_length, stop_k, block_len, resamples, seed, a.from, a.to))
}

/// Convert `GeneralizeCmd` into the resolved argument shape the generalize sugar
/// consumes. `--real` is a `>=2`-distinct comma list; the stop knobs are single
/// values, each OPTIONAL (#217) and independently defaulting to the single-
/// sourced [`R_SMA_STOP_LENGTH`]/[`R_SMA_STOP_K`] regime when omitted — clap's
/// typed `Option<i64>`/`Option<f64>` already forbids a multi-value spelling, so
/// no separate regime refusal is needed here. Every refusal whose flag survives
/// #220 reuses the old message string (byte-identical front-end).
#[allow(clippy::type_complexity)]
fn generalize_args_from(
    a: &GeneralizeCmd,
) -> Result<(String, Vec<String>, i64, f64, String, Option<i64>, Option<i64>), String> {
    let symbols: Vec<String> = match a.real.as_deref() {
        None => return Err("generalize requires --real <SYM1,SYM2,...> — a comma list of two or more instruments".to_string()),
        Some(v) => {
            let parts: Vec<String> = v.split(',').map(|s| s.to_string()).collect();
            if parts.iter().any(|s| s.is_empty()) {
                return Err("generalize: --real takes a comma list of non-empty symbols (e.g. GER40,USDJPY)".to_string());
            }
            parts
        }
    };
    if symbols.len() < 2 {
        return Err(format!(
            "generalize needs at least two instruments to compare across; got {} (--real takes a comma list of >=2 symbols)",
            symbols.len()
        ));
    }
    let mut seen = HashSet::new();
    if !symbols.iter().all(|s| seen.insert(s.clone())) {
        return Err("generalize: each instrument may appear once; --real has a duplicate symbol".to_string());
    }
    let stop_length = a.stop_length.unwrap_or(R_SMA_STOP_LENGTH);
    let stop_k = a.stop_k.unwrap_or(R_SMA_STOP_K);
    let metric = a.metric.clone().unwrap_or_else(|| "expectancy_r".to_string());
    let name = a.name.clone().unwrap_or_else(|| "generalize".to_string());
    Ok((name, symbols, stop_length, stop_k, metric, a.from, a.to))
}

/// The shared `--real`/`--from`/`--to` resolution for the family subcommands: a
/// non-empty symbol yields `DataChoice::Real`, a window flag without `--real` is a
/// usage error, absence is synthetic. Single-sourced (the old `RealWindowGrammar`
/// finish logic) so the family subcommands agree.
fn data_choice_from(
    real: Option<&str>,
    from: Option<i64>,
    to: Option<i64>,
    usage: &impl Fn() -> String,
) -> Result<DataChoice, String> {
    match real {
        Some("") => Err(usage()),
        Some(s) => Ok(DataChoice::Real { symbol: s.to_string(), from_ms: from, to_ms: to }),
        None if from.is_some() || to.is_some() => Err(usage()),
        None => Ok(DataChoice::Synthetic),
    }
}

/// Resolve `--name`/`--trace` (mutually exclusive) into `(family_name, persist)`,
/// defaulting the name when neither is given.
fn name_persist(
    name: Option<&str>,
    trace: Option<&str>,
    default: &str,
    usage: &impl Fn() -> String,
) -> Result<(String, bool), String> {
    match (name, trace) {
        (Some(_), Some(_)) => Err(usage()),
        (Some(n), None) => Ok((n.to_string(), false)),
        (None, Some(t)) => Ok((t.to_string(), true)),
        (None, None) => Ok((default.to_string(), false)),
    }
}

/// Parse the repeatable `--axis <name>=<csv>` list into by-name grid axes, mirroring
/// the old blueprint-sweep axis grammar: an empty/duplicate name or a malformed csv
/// is a usage error.
fn parse_axes(
    raw: &[String],
    usage: &impl Fn() -> String,
) -> Result<Vec<(String, Vec<Scalar>)>, String> {
    let mut axes: Vec<(String, Vec<Scalar>)> = Vec::new();
    for item in raw {
        let (n, csv) = item.split_once('=').ok_or_else(usage)?;
        if n.is_empty() || axes.iter().any(|(a, _)| a == n) {
            return Err(usage());
        }
        let vals = parse_scalar_csv(csv).ok_or_else(usage)?;
        axes.push((n.to_string(), vals));
    }
    Ok(axes)
}

/// The three distinguishable ways `validate_and_register_axes` can fail: an
/// unknown axis name is a usage error (exit 2, echoed before the archive is
/// touched); a registry write failure is a runtime error (exit 1); running
/// outside a project (no `Aura.toml` found up from cwd, #218's gate) is also
/// a runtime error (exit 1) — the strategy document cannot resolve against a
/// project store/vocabulary that doesn't exist.
enum AxisRegisterError {
    UnknownAxis(String),
    Registry(String),
    NoProject(String),
}

/// Validate every `--axis` name against the blueprint's WRAPPED probe namespace
/// (`blueprint_axis_probe`/`--list-axes`) — an axis naming a BOUND param (#246:
/// re-openable, same as every already-open knob) passes exactly like an open
/// one; only a name matching NEITHER space is refused — then canonicalize +
/// register the blueprint by topology hash and strip every axis name to the RAW
/// campaign namespace (the sweep sequence, #210 c0110). A raw-form or
/// fat-fingered axis name is refused here, echoing exactly what the user typed,
/// before the archive is touched or the blueprint is registered — shared by
/// every campaign-path dispatcher (sweep/generalize/walkforward/mc all landed
/// the identical block; #220 slice-1 deferred this dedup to "once wf/mc land
/// the same block", rule-of-three now exceeded 4x). The override set itself is
/// re-derived downstream (`CliMemberRunner::run_member`,
/// `verb_sugar::validate_before_register`) rather than threaded through this
/// return value — this preflight only decides go/no-go on the NAME.
#[allow(clippy::type_complexity)]
fn validate_and_register_axes(
    verb: &str,
    doc: &str,
    axes: &[(String, Vec<Scalar>)],
    env: &project::Env,
) -> Result<(String, Vec<(String, Vec<Scalar>)>), AxisRegisterError> {
    let space = blueprint_axis_probe(doc, env).param_space();
    let blueprint = blueprint_from_json(doc, &|t| env.resolve(t))
        .expect("doc parse-validated at the dispatch boundary");
    let bound = wrapped_bound_names(&blueprint);
    for (n, _) in axes {
        if !space.iter().any(|p| &p.name == n) && !bound.contains(n) {
            return Err(AxisRegisterError::UnknownAxis(format!(
                "axis \"{n}\" is not one of this blueprint's \
                 sweepable axes — run 'aura sweep <bp> --list-axes' \
                 to see them"
            )));
        }
    }
    if env.provenance().is_none() {
        let cwd = std::env::current_dir()
            .map(|d| d.display().to_string())
            .unwrap_or_default();
        return Err(AxisRegisterError::NoProject(format!(
            "{verb} needs a project: strategies resolve against the project \
             store and vocabulary (no Aura.toml found up from {cwd})"
        )));
    }
    let canonical = blueprint_to_json(&blueprint).expect("a loaded blueprint re-serializes");
    let reg = env.registry();
    let topo = topology_hash(&blueprint);
    reg.put_blueprint(&topo, &canonical)
        .map_err(|e| AxisRegisterError::Registry(e.to_string()))?;
    let raw_axes: Vec<(String, Vec<Scalar>)> = axes
        .iter()
        .map(|(n, v)| (campaign_run::wrapped_to_raw_axis(n).to_string(), v.clone()))
        .collect();
    Ok((canonical, raw_axes))
}

/// Exit-map for a `validate_and_register_axes` failure — the stderr line and
/// exit code every campaign dispatcher (sweep/generalize/walkforward/mc)
/// preserves: an unknown axis is a usage error (exit 2, echoed before the
/// archive is touched); a registry write failure or a missing project (no
/// `Aura.toml` found up from cwd, #218's gate) is a runtime error (exit 1).
fn exit_axis_register_error(e: AxisRegisterError) -> ! {
    match e {
        AxisRegisterError::UnknownAxis(m) => {
            eprintln!("aura: {m}");
            std::process::exit(2)
        }
        AxisRegisterError::Registry(m) => {
            eprintln!("aura: {m}");
            std::process::exit(1)
        }
        AxisRegisterError::NoProject(m) => {
            eprintln!("aura: {m}");
            std::process::exit(1)
        }
    }
}

/// Terminate per a campaign-path result (#272): a String error is a refusal
/// (exit 1); an Ok carrying the failed-cell count exits 3 when any cell failed
/// ("completed with failed cells"), else returns cleanly (exit 0).
pub(crate) fn exit_on_campaign_result(r: Result<usize, String>) {
    match r {
        Err(m) => {
            eprintln!("aura: {m}");
            std::process::exit(1);
        }
        Ok(0) => {}
        Ok(_) => std::process::exit(3),
    }
}

/// The shared campaign window in Unix-ms, clipped to the archive. `full_window`
/// probes the ARCHIVE's actual first/last bar in range (never the literal ms
/// request) and returns aura's native epoch-ns `Timestamp` (the ms->ns crossing
/// happens once, at the ingest seam — C3); the campaign document's `Window`
/// field is Unix-ms (same currency as `--from`/`--to` and every existing
/// campaign fixture, e.g. `campaign_doc_json` in research_docs.rs). Convert
/// back through the seam's own `aura_ingest::epoch_ns_to_unix_ms` at this one
/// seam-crossing (never reimplement the division inline) so the executor's
/// `unix_ms_to_epoch_ns` re-normalizes exactly once downstream, not twice.
/// Shared by the campaign dispatchers — sweep/walkforward/mc always clip;
/// generalize reaches here once per listed symbol, only through its
/// no-explicit-window fallback, and intersects the per-symbol results into the
/// one shared window (#213); an explicit `--from`+`--to` pair passes through
/// unclipped there BY DESIGN.
fn campaign_window_ms(choice: DataChoice, env: &project::Env) -> (i64, i64) {
    let source = DataSource::from_choice(choice, env);
    let (from_ts, to_ts) = source.full_window(env);
    (
        aura_ingest::epoch_ns_to_unix_ms(from_ts),
        aura_ingest::epoch_ns_to_unix_ms(to_ts),
    )
}

type SymbolSpans<'a> = Vec<(&'a str, (i64, i64))>;

/// The pure decision `generalize`'s no-explicit-window fallback reduces to once
/// every listed symbol's full window is resolved (#213): the shared window is
/// the intersection (latest start, earliest end); `Err` carries each symbol
/// paired with its own resolved window when the intersection is empty (the
/// archives never overlap), for the caller's eprintln-then-exit(1) refusal.
/// Kept separate from `dispatch_generalize` so the intersect-or-refuse
/// arithmetic is unit-testable on synthetic windows — real archives on this
/// host never disjoint (every symbol's tail reaches the live present), so the
/// refusal branch has no reachable e2e fixture.
fn intersect_shared_window<'a>(
    symbols: &'a [String],
    windows: &[(i64, i64)],
) -> Result<(i64, i64), SymbolSpans<'a>> {
    let shared_from = windows.iter().map(|w| w.0).max().expect("caller passes at least one window");
    let shared_to = windows.iter().map(|w| w.1).min().expect("caller passes at least one window");
    if shared_from > shared_to {
        Err(symbols.iter().map(String::as_str).zip(windows.iter().copied()).collect())
    } else {
        Ok((shared_from, shared_to))
    }
}

/// `aura run`: the loaded-blueprint branch (an existing `.json` first-positional) or
/// the built-in harness-kind dispatch.
fn dispatch_run(a: RunCmd, env: &project::Env) {
    match is_blueprint_file(&a.blueprint) {
        Some(path) => {
            // The loaded-blueprint grammar takes only --params/--seed/--real/--from/--to;
            // the built-in-only flags are rejected here (exit 2), never silently dropped —
            // mirroring the sweep/mc blueprint branches, which reject their non-branch flags
            // exhaustively (refuse-don't-guess). clap's optional `[blueprint]` positional
            // makes these structurally parseable, so the guard is re-asserted at dispatch.
            if a.trace.is_some() {
                eprintln!("aura: Usage: aura run <blueprint.json> [--params <json-cell-array>] [--seed <n>] [--real <SYMBOL> [--from <ms>] [--to <ms>]]");
                std::process::exit(2);
            }
            let doc = std::fs::read_to_string(path).unwrap_or_else(|e| {
                eprintln!("aura: {path}: {e}");
                std::process::exit(2);
            });
            let signal = blueprint_from_json(&doc, &|t| env.resolve(t)).unwrap_or_else(|e| {
                let mut msg = graph_construct::blueprint_load_prose(&e);
                if let Some(hint) = graph_construct::unresolved_namespace_hint(&e, env) {
                    msg.push_str(" — ");
                    msg.push_str(&hint);
                }
                eprintln!("aura: {path}: {msg}");
                std::process::exit(2);
            });
            // `wrap_r` (below, via `run_signal_r` AND the axis probe right here) welds
            // the loaded signal by its hardcoded `"bias"` output port (the R-evaluator
            // contract). A signal exposing any other output shape (a measurement-only
            // blueprint) would otherwise only surface as `blueprint_axis_probe`'s
            // `wrap_r` call panicking on `BuildError::UnknownOutPort` a few lines
            // down — refuse here instead, before that probe runs.
            if !signal.output().iter().any(|o| o.name == "bias") {
                let exposed: Vec<&str> = signal.output().iter().map(|o| o.name.as_str()).collect();
                eprintln!(
                    "aura: `aura run` expects a strategy with a bias output; this blueprint \
                     exposes {}",
                    if exposed.is_empty() {
                        "no outputs".to_string()
                    } else {
                        exposed.join(", ")
                    }
                );
                std::process::exit(1);
            }
            // Refuse an open (free-knob) blueprint at the dispatch boundary, mirroring
            // `blueprint_mc_family`'s closed-guard: `run` bootstraps over the EMPTY point, so a
            // free knob would panic in `compile_with_params` — reject it clean instead (#176).
            let free = blueprint_axis_probe(&doc, env).param_space();
            if !free.is_empty() {
                eprintln!(
                    "aura: run requires a closed blueprint (no free parameters); {} free knob(s) — \
                     bind them or use `aura sweep --axis`",
                    free.len()
                );
                std::process::exit(2);
            }
            let params = match a.params.as_deref() {
                Some(j) => parse_param_cells(j).unwrap_or_else(|m| {
                    eprintln!("aura: {m}");
                    std::process::exit(2);
                }),
                None => Vec::new(),
            };
            let data = run_data_from(a.real.as_deref(), a.from, a.to);
            let report = run_signal_r(signal, &params, data, a.seed.unwrap_or(0), env);
            println!("{}", report.to_json());
        }
        None => {
            eprintln!(
                "aura: Usage: aura run <blueprint.json> [--params <json-cell-array>] \
                 [--seed <n>] [--real <SYMBOL> [--from <ms>] [--to <ms>]]"
            );
            std::process::exit(2);
        }
    }
}

fn dispatch_chart(a: ChartCmd, env: &project::Env) {
    emit_chart(
        &a.name,
        a.tap.as_deref(),
        if a.panels { ChartMode::Panels } else { ChartMode::Overlay },
        env,
    );
}

fn dispatch_graph(a: GraphCmd, env: &project::Env) {
    match a.sub {
        None => match is_blueprint_file(&a.blueprint) {
            Some(path) => {
                let doc = std::fs::read_to_string(path).unwrap_or_else(|e| {
                    eprintln!("aura: {path}: {e}");
                    std::process::exit(2);
                });
                let bp = graph_construct::composite_from_any(&doc, env).unwrap_or_else(|msg| {
                    eprintln!("aura: {path}: {msg}");
                    std::process::exit(2);
                });
                print!("{}", render::render_html(&bp));
            }
            None if a.blueprint.is_none() => {
                let bp = blueprint_from_json(
                    include_str!("../examples/r_sma.json"),
                    &|t| env.resolve(t),
                )
                .expect("the shipped r-sma example reloads into a renderable blueprint");
                print!("{}", render::render_html(&bp));
            }
            None => {
                eprintln!(
                    "aura: Usage: aura graph [<blueprint.json>] [build|introspect|register]; \
                     {} is not a readable .json blueprint",
                    a.blueprint.as_deref().unwrap_or_default()
                );
                std::process::exit(2);
            }
        },
        Some(GraphSub::Build) => graph_construct::build_cmd(env),
        Some(GraphSub::Introspect(i)) => graph_construct::introspect_cmd(i, env),
        Some(GraphSub::Register { file }) => graph_construct::register_cmd(&file, env),
    }
}

fn dispatch_generalize(a: GeneralizeCmd, env: &project::Env) {
    let usage = || format!("Usage: aura generalize <blueprint.json> --real <SYM1,SYM2[,…]> --axis <name>=<value> [--axis …] [--stop-length <n> (default {R_SMA_STOP_LENGTH})] [--stop-k <x> (default {R_SMA_STOP_K:.1})] [--metric <m>] [--name <n>] [--from <ms>] [--to <ms>]");
    let Some(path) = is_blueprint_file(&a.blueprint) else {
        eprintln!("aura: {}", usage());
        std::process::exit(2);
    };
    let doc = std::fs::read_to_string(path).unwrap_or_else(|e| {
        eprintln!("aura: {path}: {e}");
        std::process::exit(2);
    });
    if let Err(msg) = graph_construct::blueprint_slot_prose(&doc, env) {
        eprintln!("aura: {path}: {msg}");
        std::process::exit(2);
    }
    let (name, symbols, stop_length, stop_k, metric, from, to) =
        generalize_args_from(&a).unwrap_or_else(|m| {
            eprintln!("aura: {m}");
            std::process::exit(2);
        });
    // Data-free R-metric refusal, byte-identical to the retired welded path
    // (exit 2) before any archive is touched.
    if let Err(e) = check_r_metric(&metric) {
        eprintln!("aura: {e}");
        std::process::exit(2);
    }
    let axes = parse_axes(&a.axis, &usage).unwrap_or_else(|m| {
        eprintln!("aura: {m}");
        std::process::exit(2);
    });
    if axes.is_empty() {
        eprintln!("aura: {}", usage());
        std::process::exit(2);
    }
    // A candidate is a single grid cell, not a sweep: every axis carries
    // exactly one value (the refusal the retired `--fast 2,3` grammar made).
    for (n, vals) in &axes {
        if vals.len() != 1 {
            eprintln!(
                "aura: generalize: each --axis takes exactly one value; axis \"{n}\" has {}",
                vals.len()
            );
            std::process::exit(2);
        }
    }
    let (canonical, raw_axes) = validate_and_register_axes("generalize", &doc, &axes, env)
        .unwrap_or_else(|e| exit_axis_register_error(e));
    // Window: the explicit --from/--to when both present (byte-identical to the
    // retired welded path, verified by the exact-grade anchor); otherwise the
    // INTERSECTION of every listed symbol's full archive window — the only span
    // over which every instrument actually has data (#213). A single listed
    // symbol degenerates to its own full window unchanged.
    let (from_ms, to_ms) = match (from, to) {
        (Some(f), Some(t)) => (f, t),
        _ => {
            let windows: Vec<(i64, i64)> = symbols
                .iter()
                .map(|s| {
                    campaign_window_ms(
                        DataChoice::Real { symbol: s.clone(), from_ms: from, to_ms: to },
                        env,
                    )
                })
                .collect();
            match intersect_shared_window(&symbols, &windows) {
                Ok(w) => w,
                Err(spans) => {
                    for (s, w) in spans {
                        eprintln!("aura: generalize: {s} spans [{}, {}]", w.0, w.1);
                    }
                    eprintln!(
                        "aura: generalize: no window is shared across all listed symbols \
                         (their archives do not overlap)"
                    );
                    std::process::exit(1);
                }
            }
        }
    };
    let inv = verb_sugar::SugarInvocation {
        axes: &raw_axes,
        name,
        symbols,
        from_ms,
        to_ms,
        blueprint_canonical: &canonical,
        stop: Some(verb_sugar::VolStop { length: stop_length, k: stop_k }),
        // `generalize` has no `--trace` flag in its own grammar (#224 delivered
        // sweep + walkforward only); trace-writing is out of scope here.
        trace: false,
    };
    exit_on_campaign_result(verb_sugar::run_generalize_sugar(&inv, &metric, env));
}

fn dispatch_runs(a: RunsCmd, env: &project::Env) {
    match a.sub {
        RunsSub::Families => runs_families(env),
        RunsSub::Family { id, rank_kw, metric } => match (rank_kw.as_deref(), metric) {
            (None, None) => runs_family(&id, None, env),
            (Some("rank"), Some(m)) => runs_family(&id, Some(&m), env),
            _ => {
                eprintln!("aura: Usage: aura runs family <id> [rank <metric>]");
                std::process::exit(2);
            }
        },
    }
}

fn dispatch_reproduce(a: ReproduceCmd, env: &project::Env) {
    reproduce_family(&a.id, env);
}

fn dispatch_new(a: NewCmd, _env: &project::Env) {
    let cwd = std::env::current_dir().unwrap_or_else(|e| {
        eprintln!("aura: {e}");
        std::process::exit(1);
    });
    let spec = scaffold::project_scaffold_spec(&a.name, &cwd).unwrap_or_else(|m| {
        eprintln!("aura: {m}");
        std::process::exit(2);
    });
    scaffold::scaffold_project(&spec).unwrap_or_else(|m| {
        eprintln!("aura: {m}");
        std::process::exit(1);
    });
    println!(
        "created project \"{}\" (data-only; attach native nodes later with `aura nodes new`)",
        spec.name
    );
}

fn dispatch_nodes(cmd: NodesCmd) {
    match cmd.command {
        NodesCommand::New(a) => dispatch_nodes_new(a),
    }
}

fn dispatch_nodes_new(a: NodesNewCmd) {
    let cwd = std::env::current_dir().unwrap_or_else(|e| {
        eprintln!("aura: {e}");
        std::process::exit(1);
    });
    let Some(root) = project::discover_from(&cwd) else {
        eprintln!(
            "aura: `aura nodes new` needs a project (no Aura.toml found up from {})",
            cwd.display()
        );
        std::process::exit(1);
    };
    let toml = match project::read_aura_toml(&root) {
        Ok(t) => t,
        Err(e) => {
            eprintln!("aura: {e}");
            std::process::exit(1);
        }
    };
    // Mirror `append_nodes_pointer`'s textual `[nodes]`-section check (not just
    // `crates.is_empty()`) so a `[nodes]` section with an empty crates array
    // is caught here too, before scaffold_node_crate writes a crate that
    // append_nodes_pointer would then refuse to attach.
    let already_attached = !toml.nodes.crates.is_empty()
        || std::fs::read_to_string(root.join("Aura.toml")).is_ok_and(|t| t.contains("[nodes]"));
    if already_attached {
        eprintln!("aura: a node crate is already attached (multi-crate loading is not yet supported)");
        std::process::exit(1);
    }
    let parent = root.parent().unwrap_or(&root).to_path_buf();
    let spec = match scaffold::scaffold_spec(
        &a.name,
        a.engine_path.as_deref(),
        a.namespace.as_deref(),
        &parent,
    ) {
        Ok(s) => s,
        Err(m) => {
            eprintln!("aura: {m}");
            std::process::exit(2);
        }
    };
    if let Err(m) = scaffold::scaffold_node_crate(&spec) {
        eprintln!("aura: {m}");
        std::process::exit(1);
    }
    if let Err(m) = scaffold::append_nodes_pointer(&root, &format!("../{}", a.name)) {
        eprintln!("aura: {m}");
        std::process::exit(1);
    }
    println!(
        "created node crate \"../{}\" (namespace \"{}\") and attached it to {}",
        a.name,
        spec.namespace,
        root.file_name().map(|s| s.to_string_lossy().into_owned()).unwrap_or_default()
    );
}

fn dispatch_data(cmd: DataCmd, env: &project::Env) {
    match cmd.command {
        DataCommand::Coverage(a) => dispatch_data_coverage(a, env),
        DataCommand::List => dispatch_data_list(env),
    }
}

/// `aura data list` (#264 cut 2): print the archive's known symbols, sorted
/// ascending, one per line — the discovery step before `aura data coverage`
/// or scoping a campaign's instrument matrix. Resolves the archive root
/// through the normal project path ([`dispatch_data_coverage`]'s
/// `Env::data_path`), then reuses `DataServer`'s own symbol index
/// (`DataServer::symbols()`, already sorted) rather than re-deriving a
/// directory scan. An empty or absent archive is informational absence, not a
/// fault: a `no symbols` prose line, exit 0 (the verb corpus's established
/// empty-result register — cf. `runs family`'s unknown-id empty exit 0).
fn dispatch_data_list(env: &project::Env) {
    let data_path = std::path::PathBuf::from(env.data_path());
    let server = aura_ingest::DataServer::new(&data_path);
    for line in data_list_report(&server.symbols()) {
        println!("{line}");
    }
}

/// Pure: render `symbols` (already sorted by `DataServer::symbols()`) as one
/// line per symbol, or a single `no symbols` line when the archive is empty
/// or absent — informational absence, never a fault (#264).
fn data_list_report(symbols: &[std::sync::Arc<str>]) -> Vec<String> {
    if symbols.is_empty() {
        return vec!["no symbols".to_string()];
    }
    symbols.iter().map(|s| s.to_string()).collect()
}

/// `aura data coverage <SYMBOL>` (#264 cut 1): print the archive's month
/// coverage for `SYMBOL` — the Copper failure mode (files present at both
/// ends of a window while an interior month is missing, so a first/last-bounds
/// check passes but a campaign run aborts mid-window) made visible up front.
/// Resolves the archive root through the normal project path (`Env::data_path`
/// — a project-local `[paths] data` override when set, the data-server default
/// otherwise), so a project-scoped archive is inventoried, not the host one.
fn dispatch_data_coverage(a: DataCoverageCmd, env: &project::Env) {
    let data_path = std::path::PathBuf::from(env.data_path());
    let months = aura_ingest::list_m1_months(&data_path, &a.symbol);
    match data_coverage_report(&a.symbol, &months) {
        Ok(lines) => {
            for line in lines {
                println!("{line}");
            }
        }
        Err(e) => {
            eprintln!("aura: {e} at {}", env.data_path());
            std::process::exit(1);
        }
    }
}

/// Pure: render `symbol`'s coverage report from its sorted `(year, month)` file
/// list — one `span:` line framing the first/last present month, then either a
/// `no gaps` line or one `missing: YYYY-MM..YYYY-MM` line per interior
/// contiguous gap (a ten-month hole is one line, not ten). `Err` exactly when
/// `months` is empty — no archive file exists for `symbol` at all (the
/// caller's "unknown symbol" refusal, stderr + exit 1).
fn data_coverage_report(symbol: &str, months: &[(u16, u8)]) -> Result<Vec<String>, String> {
    let Some(&first) = months.first() else {
        return Err(format!("no archive files found for symbol \"{symbol}\""));
    };
    let last = *months.last().expect("non-empty checked above");
    let mut lines = vec![format!("{symbol} span: {}..{}", fmt_year_month(first), fmt_year_month(last))];
    let gaps: Vec<((u16, u8), (u16, u8))> = months
        .windows(2)
        .filter_map(|pair| {
            let (prev, next) = (pair[0], pair[1]);
            let expected = next_year_month(prev);
            (expected != next).then(|| (expected, prev_year_month(next)))
        })
        .collect();
    if gaps.is_empty() {
        lines.push(format!("{symbol} no gaps"));
    } else {
        for (from, to) in gaps {
            lines.push(format!("{symbol} missing: {}..{}", fmt_year_month(from), fmt_year_month(to)));
        }
    }
    Ok(lines)
}

fn fmt_year_month((y, m): (u16, u8)) -> String {
    format!("{y:04}-{m:02}")
}

fn next_year_month((y, m): (u16, u8)) -> (u16, u8) {
    if m == 12 { (y + 1, 1) } else { (y, m + 1) }
}

fn prev_year_month((y, m): (u16, u8)) -> (u16, u8) {
    if m == 1 { (y - 1, 12) } else { (y, m - 1) }
}

/// `aura sweep`: loaded-blueprint by-name axis sweep (or `--list-axes` probe) when the
/// first-positional is an existing `.json`, else the built-in `--strategy` grid sweep.
fn dispatch_sweep(a: SweepCmd, env: &project::Env) {
    // Single-sourced: the blueprint grammar and the no-blueprint usage error must
    // stay in lockstep, so both arms below read this one closure.
    let usage = || "Usage: aura sweep <blueprint.json> --axis <name>=<csv> [--axis …] [--name <n> | --trace <n>] [--real <SYM> [--from <ms>] [--to <ms>]]".to_string();
    match is_blueprint_file(&a.blueprint) {
        Some(path) => {
            let doc = std::fs::read_to_string(path).unwrap_or_else(|e| {
                eprintln!("aura: {path}: {e}");
                std::process::exit(2);
            });
            // Parse-validate the blueprint once at the boundary (with file-path context),
            // house-style prose (#184's convention, single-sourced — #210 c0110 finding).
            if let Err(msg) = graph_construct::blueprint_slot_prose(&doc, env) {
                eprintln!("aura: {path}: {msg}");
                std::process::exit(2);
            }
            // A built-in-only flag with a blueprint file is not in this grammar.
            if a.strategy.is_some() {
                eprintln!("aura: {}", usage());
                std::process::exit(2);
            }
            if a.list_axes {
                // A query, not a sweep: it must stand alone.
                if !a.axis.is_empty()
                    || a.name.is_some()
                    || a.trace.is_some()
                    || a.real.is_some()
                    || a.from.is_some()
                    || a.to.is_some()
                {
                    eprintln!("aura: --list-axes lists axes and takes no other flags");
                    std::process::exit(2);
                }
                list_blueprint_axes(&doc, env);
                return;
            }
            let axes = parse_axes(&a.axis, &usage).unwrap_or_else(|m| {
                eprintln!("aura: {m}");
                std::process::exit(2);
            });
            if axes.is_empty() {
                eprintln!("aura: {}", usage());
                std::process::exit(2);
            }
            let (name, persist) = name_persist(a.name.as_deref(), a.trace.as_deref(), "sweep", &usage)
                .unwrap_or_else(|m| {
                    eprintln!("aura: {m}");
                    std::process::exit(2);
                });
            let data = data_choice_from(a.real.as_deref(), a.from, a.to, &usage).unwrap_or_else(|m| {
                eprintln!("aura: {m}");
                std::process::exit(2);
            });
            match &data {
                DataChoice::Real { symbol, .. } => {
                    // The dissolved branch: real-data blueprint sweeps run as
                    // sugar over a generated campaign document through the
                    // one campaign executor. The blueprint store write (by
                    // topology hash) is kept — the
                    // canonical bytes are also the strategy ref's content, and
                    // `topology_hash` IS `content_id_of` applied to the same
                    // canonical bytes (verified by reading: `topology_hash` ==
                    // `content_id` == `aura_research::content_id_of`), so the
                    // strategy ref resolves against this one write; a second
                    // put under `content_id_of(&canonical)` would target the
                    // identical key and is not needed.
                    // Axis validation + canonicalize/register + the wrapped->raw
                    // strip are single-sourced in `validate_and_register_axes`
                    // (data-free, so this fires before the archive is touched).
                    let (canonical, raw_axes) = validate_and_register_axes("sweep", &doc, &axes, env)
                        .unwrap_or_else(|e| exit_axis_register_error(e));
                    let symbol = symbol.clone();
                    // Archive-clipped Unix-ms window; the ns->ms seam-crossing
                    // rationale lives on `campaign_window_ms`.
                    let (from_ms, to_ms) = campaign_window_ms(data, env);
                    let inv = verb_sugar::SugarInvocation {
                        axes: &raw_axes,
                        name: name.clone(),
                        symbols: vec![symbol.clone()],
                        from_ms,
                        to_ms,
                        blueprint_canonical: &canonical,
                        stop: None,
                        // #224: the real-data campaign path DELIVERS `--trace`
                        // (per-member tap-series persistence); `persist` is
                        // `name_persist`'s bool (true iff `--trace` was given).
                        trace: persist,
                    };
                    exit_on_campaign_result(verb_sugar::run_sweep_sugar(&inv, env));
                }
                DataChoice::Synthetic => {
                    // The synthetic in-process family path is still reduce-only
                    // (`blueprint_sweep_family` writes no per-member traces, #224
                    // delivered only the real-data campaign path) — refuse rather
                    // than silently accept an advertised-but-unhonoured flag.
                    if persist {
                        eprintln!(
                            "aura: --trace is not yet available on a synthetic sweep (no --real); see #224"
                        );
                        std::process::exit(2);
                    }
                    run_blueprint_sweep(
                        &doc, &axes, &name, persist,
                        DataSource::from_choice(data, env), env,
                    );
                }
            }
        }
        None => {
            eprintln!("aura: {}", usage());
            std::process::exit(2);
        }
    }
}

/// `aura walkforward`: IS-refit walk-forward over a loaded blueprint — the
/// synthetic in-process family (no `--real`) or the campaign path (`--real`,
/// #220: arbitrary user blueprint + axes, formerly the welded r-sma branch).
fn dispatch_walkforward(a: WalkforwardCmd, env: &project::Env) {
    // Single-sourced: both arms below read this one closure (house style, #179).
    let usage = || format!("Usage: aura walkforward <blueprint.json> --axis <name>=<csv> [--axis …] [--select <argmax|plateau:mean|plateau:worst>] [--name <n>] | aura walkforward <blueprint.json> --real <SYMBOL> --axis <name>=<csv> [--axis …] [--stop-length <n> (default {R_SMA_STOP_LENGTH})] [--stop-k <x> (default {R_SMA_STOP_K:.1})] [--from <ms>] [--to <ms>] [--name <n> | --trace <n>]");
    match is_blueprint_file(&a.blueprint) {
        Some(path) => {
            let doc = std::fs::read_to_string(path).unwrap_or_else(|e| {
                eprintln!("aura: {path}: {e}");
                std::process::exit(2);
            });
            if let Err(msg) = graph_construct::blueprint_slot_prose(&doc, env) {
                eprintln!("aura: {path}: {msg}");
                std::process::exit(2);
            }
            let axes = parse_axes(&a.axis, &usage).unwrap_or_else(|m| {
                eprintln!("aura: {m}");
                std::process::exit(2);
            });
            if axes.is_empty() {
                eprintln!("aura: walkforward <blueprint.json> requires >= 1 --axis to re-fit per window");
                std::process::exit(2);
            }
            if a.real.is_some() {
                // The campaign path (#220): the real-archive execution routes
                // through the one campaign executor, over the user's own
                // blueprint and axes (formerly the welded r-sma branch).
                let select = match a.select.as_deref() {
                    Some(s) => match parse_select(s) {
                        Ok(sel) => select_rule_of(sel),
                        Err(()) => {
                            eprintln!("aura: {}", usage());
                            std::process::exit(2);
                        }
                    },
                    None => aura_research::SelectRule::Argmax,
                };
                let (name, trace, symbol, stop_length, stop_k, from, to) =
                    walkforward_args_from(&a).unwrap_or_else(|m| {
                        eprintln!("aura: {m}");
                        std::process::exit(2);
                    });
                // Axis validation + canonicalize/register + the wrapped->raw
                // strip are single-sourced in `validate_and_register_axes`
                // (data-free, so this fires before the archive is touched).
                let (canonical, raw_axes) = validate_and_register_axes("walkforward", &doc, &axes, env)
                    .unwrap_or_else(|e| exit_axis_register_error(e));
                // Unlike `dispatch_generalize` (a single run per instrument, insensitive
                // to a day's edge shift), `blueprint_walkforward_family` sources its span
                // from `DataSource::wf_full_span`, which — for Real — ALWAYS clips
                // `--from`/`--to` to the archive's actual first/last bar in range (never the
                // literal ms request): a holiday/weekend edge shifts every IS/OOS window's
                // calendar placement, so the roller must clip identically here too, even
                // when both flags are given, or the per-window winners and OOS pips diverge
                // from the committed exact-grade anchor.
                let (from_ms, to_ms) = campaign_window_ms(
                    DataChoice::Real { symbol: symbol.clone(), from_ms: from, to_ms: to },
                    env,
                );
                let (is_ms, oos_ms, step_ms) = fit_wf_ms_sizes(from_ms, to_ms);
                let inv = verb_sugar::SugarInvocation {
                    axes: &raw_axes,
                    name,
                    symbols: vec![symbol],
                    from_ms,
                    to_ms,
                    blueprint_canonical: &canonical,
                    stop: Some(verb_sugar::VolStop { length: stop_length, k: stop_k }),
                    // #224: the walkforward sibling delivers `--trace` identically
                    // to sweep, riding the SAME per-cell nominee trace mechanism
                    // (unchanged since 0109 — walkforward always nominates).
                    trace,
                };
                let result = verb_sugar::run_walkforward_sugar(
                    &inv,
                    WINNER_SELECTION_METRIC,
                    verb_sugar::WfWindows {
                        in_sample_ms: is_ms,
                        out_of_sample_ms: oos_ms,
                        step_ms,
                    },
                    select,
                    env,
                );
                exit_on_campaign_result(result);
                return;
            }
            // Synthetic in-process family path — unchanged (#220 non-goal); it
            // persists no per-window taps, so `--trace` stays refused here (#224
            // delivered only the `--real` campaign path above), mirroring
            // `dispatch_sweep`'s synthetic-arm refusal with its own named pointer.
            if a.trace.is_some() {
                eprintln!(
                    "aura: --trace is not yet available on a synthetic walkforward (no --real); see #224"
                );
                std::process::exit(2);
            }
            if a.stop_length.is_some() || a.stop_k.is_some() {
                eprintln!("aura: {}", usage());
                std::process::exit(2);
            }
            let select = match a.select.as_deref() {
                Some(s) => parse_select(s).unwrap_or_else(|()| {
                    eprintln!("aura: {}", usage());
                    std::process::exit(2);
                }),
                None => Selection::Argmax,
            };
            let name = a.name.clone().unwrap_or_else(|| "walkforward".to_string());
            run_blueprint_walkforward(&doc, &axes, &name, DataSource::Synthetic, select, env);
        }
        None => {
            eprintln!("aura: {}", usage());
            std::process::exit(2);
        }
    }
}

/// `aura mc`: Monte-Carlo over a loaded blueprint — the synthetic seed family
/// (`--seeds`, closed blueprint) or the R-bootstrap campaign path (`--real`,
/// #220: arbitrary user blueprint + axes, formerly the welded r-sma branch).
fn dispatch_mc(a: McCmd, env: &project::Env) {
    // Single-sourced: every arm below reads this one closure (house style, #179).
    let usage = || format!("Usage: aura mc <blueprint.json> --seeds <n> [--name <n>] | aura mc <blueprint.json> --real <SYMBOL> --axis <name>=<csv> [--axis …] [--stop-length <n> (default {R_SMA_STOP_LENGTH})] [--stop-k <x> (default {R_SMA_STOP_K:.1})] [--block-len <n>] [--resamples <n>] [--seed <n>] [--from <ms>] [--to <ms>]");
    match is_blueprint_file(&a.blueprint) {
        Some(path) => {
            let doc = std::fs::read_to_string(path).unwrap_or_else(|e| {
                eprintln!("aura: {path}: {e}");
                std::process::exit(2);
            });
            if let Err(msg) = graph_construct::blueprint_slot_prose(&doc, env) {
                eprintln!("aura: {path}: {msg}");
                std::process::exit(2);
            }
            if a.real.is_some() {
                // The campaign path (#220): blueprint + --real + --axis routes
                // the R-bootstrap pipeline through the one campaign executor.
                // The two mc modes stay disjoint: --seeds belongs to the
                // synthetic seed family only.
                if a.seeds.is_some() {
                    eprintln!("aura: {}", usage());
                    std::process::exit(2);
                }
                let (name, symbol, stop_length, stop_k, block_len, resamples, seed, from, to) =
                    mc_args_from(&a).unwrap_or_else(|m| {
                        eprintln!("aura: {m}");
                        std::process::exit(2);
                    });
                let axes = parse_axes(&a.axis, &usage).unwrap_or_else(|m| {
                    eprintln!("aura: {m}");
                    std::process::exit(2);
                });
                if axes.is_empty() {
                    eprintln!("aura: {}", usage());
                    std::process::exit(2);
                }
                // Axis validation + canonicalize/register + the wrapped->raw
                // strip are single-sourced in `validate_and_register_axes`
                // (data-free, so this fires before the archive is touched).
                let (canonical, raw_axes) = validate_and_register_axes("mc", &doc, &axes, env)
                    .unwrap_or_else(|e| exit_axis_register_error(e));
                // `blueprint_walkforward_family` sources its span from
                // `DataSource::wf_full_span`, which for Real ALWAYS clips `--from`/`--to`
                // to the archive's actual first/last bar in range: a holiday/weekend edge
                // shifts every IS/OOS window's calendar placement, so the roller must clip
                // identically here too, even when both flags are given, or the per-window
                // winners and pooled OOS series diverge from the committed exact-grade anchor.
                let (from_ms, to_ms) = campaign_window_ms(
                    DataChoice::Real { symbol: symbol.clone(), from_ms: from, to_ms: to },
                    env,
                );
                let (is_ms, oos_ms, step_ms) = fit_wf_ms_sizes(from_ms, to_ms);
                let inv = verb_sugar::SugarInvocation {
                    axes: &raw_axes,
                    name,
                    symbols: vec![symbol],
                    from_ms,
                    to_ms,
                    blueprint_canonical: &canonical,
                    stop: Some(verb_sugar::VolStop { length: stop_length, k: stop_k }),
                    // mc --real refuses --name/--trace up front (`mc_args_from`) —
                    // the R-bootstrap records without a family name; trace-writing
                    // is out of #224's scope here.
                    trace: false,
                };
                let result = verb_sugar::run_mc_sugar(
                    &inv,
                    WINNER_SELECTION_METRIC,
                    verb_sugar::WfWindows {
                        in_sample_ms: is_ms,
                        out_of_sample_ms: oos_ms,
                        step_ms,
                    },
                    verb_sugar::McKnobs { resamples, block_len, seed },
                    env,
                );
                exit_on_campaign_result(result);
                return;
            }
            // Synthetic seed family (unchanged, #220 non-goal): closed
            // blueprint, --seeds only; every campaign-mode flag is rejected.
            if a.from.is_some()
                || a.to.is_some()
                || a.stop_length.is_some()
                || a.stop_k.is_some()
                || a.block_len.is_some()
                || a.resamples.is_some()
                || a.seed.is_some()
                || a.trace.is_some()
                || !a.axis.is_empty()
            {
                eprintln!("aura: {}", usage());
                std::process::exit(2);
            }
            let n_seeds = match a.seeds {
                Some(n) if n > 0 => n,
                _ => {
                    eprintln!("aura: {}", usage());
                    std::process::exit(2);
                }
            };
            let name = a.name.clone().unwrap_or_else(|| "mc".to_string());
            run_blueprint_mc(&doc, n_seeds, &name, DataSource::Synthetic, env);
        }
        None => {
            eprintln!("aura: {}", usage());
            std::process::exit(2);
        }
    }
}

fn main() {
    // Restore the default SIGPIPE disposition. Rust's runtime sets SIGPIPE to SIG_IGN
    // at startup, so a write to a closed stdout pipe (`aura sweep | head`, a closed UI
    // pane) returns EPIPE and panics in `println!` instead of terminating quietly on
    // SIGPIPE — the conventional Unix CLI behaviour. One reset covers every
    // family-emitting subcommand at once.
    #[cfg(unix)]
    unsafe {
        libc::signal(libc::SIGPIPE, libc::SIG_DFL);
    }
    let cli = Cli::parse();
    // `aura new`/`aura nodes new` scaffold; they must not require a
    // loadable project even when invoked inside one (e.g. an unbuilt tree).
    let env = if matches!(cli.command, Command::New(_) | Command::Nodes(_)) {
        project::Env::std()
    } else {
        match std::env::current_dir()
            .ok()
            .and_then(|d| project::discover_from(&d))
        {
            Some(root) => match project::load(&root, cli.release) {
                Ok(p) => project::Env::with_project(p),
                Err(e) => {
                    eprintln!("aura: {e}");
                    std::process::exit(1);
                }
            },
            None => project::Env::std(),
        }
    };
    match cli.command {
        Command::Run(a) => dispatch_run(a, &env),
        Command::Chart(a) => dispatch_chart(a, &env),
        Command::Graph(a) => dispatch_graph(a, &env),
        Command::Sweep(a) => dispatch_sweep(a, &env),
        Command::Walkforward(a) => dispatch_walkforward(a, &env),
        Command::Generalize(a) => dispatch_generalize(a, &env),
        Command::Mc(a) => dispatch_mc(a, &env),
        Command::Runs(a) => dispatch_runs(a, &env),
        Command::Reproduce(a) => dispatch_reproduce(a, &env),
        Command::New(a) => dispatch_new(a, &env),
        Command::Nodes(a) => dispatch_nodes(a),
        Command::Process(a) => research_docs::process_cmd(a, &env),
        Command::Campaign(a) => research_docs::campaign_cmd(a, &env),
        Command::Data(a) => dispatch_data(a, &env),
    }
}

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

    /// A single listed symbol degenerates to its own full window unchanged
    /// (#213) — the intersection of one window with itself is that window.
    #[test]
    fn intersect_shared_window_of_a_single_symbol_is_its_own_window() {
        let symbols = vec!["GER40".to_string()];
        let windows = vec![(100, 200)];
        assert_eq!(intersect_shared_window(&symbols, &windows), Ok((100, 200)));
    }

    /// Overlapping multi-symbol windows resolve to the intersection — the
    /// latest start, earliest end (#213) — never `symbols[0]`'s own window.
    #[test]
    fn intersect_shared_window_of_overlapping_windows_is_latest_start_earliest_end() {
        let symbols = vec!["AAPL.US".to_string(), "GER40".to_string()];
        let windows = vec![(0, 300), (100, 200)];
        assert_eq!(intersect_shared_window(&symbols, &windows), Ok((100, 200)));
    }

    /// Audit follow-up to #247/#269: `render_bind_error` is exhaustive — the
    /// axis-usage variants the old catch-all Debug-framed render as prose
    /// naming the axis, with no Rust identifier on the user's stderr.
    #[test]
    fn render_bind_error_prose_covers_the_axis_usage_variants() {
        let dup = render_bind_error(&aura_engine::BindError::DuplicateBinding("a.b".into()));
        assert!(dup.contains("a.b") && dup.contains("bound twice"), "{dup}");
        assert!(!dup.contains("DuplicateBinding"), "Debug leak: {dup}");
        let empty = render_bind_error(&aura_engine::BindError::EmptyAxis("a.b".into()));
        assert!(empty.contains("a.b") && empty.contains("no values"), "{empty}");
        assert!(!empty.contains("EmptyAxis"), "Debug leak: {empty}");
        let range = render_bind_error(&aura_engine::BindError::EmptyRange("a.b".into()));
        assert!(range.contains("a.b") && !range.contains("EmptyRange"), "Debug leak: {range}");
    }

    /// Audit follow-up to #247/#269: a `Compile` fault at the sweep boundary —
    /// a blueprint defect, not an axis usage error — renders as a prose frame
    /// that names the situation and labels the embedded compile detail as
    /// internal, instead of leaking the bare Debug struct as the whole message.
    #[test]
    fn render_bind_error_frames_a_compile_fault_as_prose_with_labeled_detail() {
        let msg = render_bind_error(&aura_engine::BindError::Compile(
            aura_engine::CompileError::BadInteriorIndex,
        ));
        assert!(msg.contains("failed to bootstrap"), "{msg}");
        assert!(msg.contains("internal detail"), "{msg}");
        assert!(!msg.starts_with("Compile"), "the frame must lead with prose: {msg}");
    }

    /// An unknown symbol (no archive files at all — an empty month list)
    /// refuses rather than reporting a bogus empty-span coverage (#264): the
    /// caller eprintln's the message and exits 1.
    #[test]
    fn data_coverage_report_refuses_an_unknown_symbol() {
        let err = data_coverage_report("GHOST", &[]).unwrap_err();
        assert!(err.contains("GHOST"), "names the unknown symbol: {err}");
    }

    /// A symbol with a fully contiguous file index reports its span plus an
    /// explicit `no gaps` line — never a bare span with no gap-status line at
    /// all, which would leave "no gaps" indistinguishable from "gaps not yet
    /// checked" (#264).
    #[test]
    fn data_coverage_report_of_a_gapless_symbol_is_span_plus_no_gaps() {
        let months = [(2024, 1), (2024, 2), (2024, 3)];
        let lines = data_coverage_report("SYMA", &months).expect("known symbol");
        assert_eq!(lines, vec!["SYMA span: 2024-01..2024-03", "SYMA no gaps"]);
    }

    /// The Copper failure shape itself: one interior gap collapses to a single
    /// `missing: YYYY-MM..YYYY-MM` line naming the whole contiguous hole, not
    /// one line per missing month (#264).
    #[test]
    fn data_coverage_report_collapses_an_interior_gap_to_one_range_line() {
        let months = [(2024, 1), (2024, 2), (2024, 5), (2024, 6)];
        let lines = data_coverage_report("GAPSYM", &months).expect("known symbol");
        assert_eq!(
            lines,
            vec!["GAPSYM span: 2024-01..2024-06", "GAPSYM missing: 2024-03..2024-04"]
        );
    }

    /// An empty (or absent) archive is informational absence, not a fault
    /// (#264): `data_list_report` returns the single `no symbols` prose line
    /// rather than an empty `Vec`, so the caller's exit-0 println always has
    /// something to say.
    #[test]
    fn data_list_report_of_an_empty_archive_is_a_no_symbols_line() {
        let symbols: Vec<std::sync::Arc<str>> = vec![];
        assert_eq!(data_list_report(&symbols), vec!["no symbols".to_string()]);
    }

    /// A non-empty archive renders one line per symbol, in the order handed
    /// in (`DataServer::symbols()` already sorts) — no `no symbols` line
    /// mixed in (#264).
    #[test]
    fn data_list_report_of_a_populated_archive_is_one_line_per_symbol() {
        let symbols: Vec<std::sync::Arc<str>> =
            vec![std::sync::Arc::from("SYMA"), std::sync::Arc::from("SYMB")];
        assert_eq!(data_list_report(&symbols), vec!["SYMA".to_string(), "SYMB".to_string()]);
    }

    /// Disjoint archives (no shared instant across all listed symbols) refuse
    /// rather than silently pooling a floor over different periods per
    /// instrument (#213); the error names each symbol next to its own span,
    /// which the caller eprintln's before exiting 1 — the boundary this
    /// covers is otherwise unreachable in an e2e fixture on this host, since
    /// every archived symbol's tail reaches the live present (no two windows
    /// are ever genuinely disjoint here).
    #[test]
    fn intersect_shared_window_of_disjoint_windows_refuses_with_every_span() {
        let symbols = vec!["A".to_string(), "B".to_string()];
        let windows = vec![(0, 100), (200, 300)];
        assert_eq!(
            intersect_shared_window(&symbols, &windows),
            Err(vec![("A", (0, 100)), ("B", (200, 300))]),
        );
    }

    /// Regenerates the shipped r_breakout examples from the carved signal. Run by hand:
    /// `cargo test --bin aura emit_r_breakout_examples -- --ignored`. The examples are
    /// green-by-construction (serialised from the builder, never hand-authored).
    #[test]
    #[ignore = "regenerates crates/aura-cli/examples/r_breakout{,_open}.json; run by hand"]
    fn emit_r_breakout_examples() {
        std::fs::create_dir_all("examples").expect("examples dir");
        std::fs::write(
            "examples/r_breakout.json",
            blueprint_to_json(&r_breakout_signal(Some(R_BREAKOUT_CHANNEL))).expect("serialize closed r_breakout"),
        )
        .expect("write examples/r_breakout.json");
        std::fs::write(
            "tests/fixtures/r_breakout_open.json",
            blueprint_to_json(&r_breakout_signal(None)).expect("serialize open r_breakout"),
        )
        .expect("write tests/fixtures/r_breakout_open.json");
    }

    /// The shipped examples are a faithful serialisation of the carved signal (#159 cut 2):
    /// re-serialising the carve equals the checked-in bytes. Green-by-construction with the
    /// emitter; a drift between builder and file breaks it. Survives the fused builder's
    /// retirement (it references `r_breakout_signal`, the carve, not the retired builder).
    #[test]
    fn shipped_r_breakout_examples_serialize_the_carved_signal() {
        assert_eq!(
            blueprint_to_json(&r_breakout_signal(Some(R_BREAKOUT_CHANNEL))).expect("serialize closed"),
            include_str!("../examples/r_breakout.json"),
        );
        assert_eq!(
            blueprint_to_json(&r_breakout_signal(None)).expect("serialize open"),
            include_str!("../tests/fixtures/r_breakout_open.json"),
        );
    }

    /// The shipped closed example, reloaded through the data plane and run, grades
    /// bit-identically to running the carved signal directly (#159 cut 2). This is
    /// r_breakout's durable equivalence anchor after the fused builder retires: it pins
    /// that `examples/r_breakout.json` still produces the carve's grade, with no
    /// hardcoded golden. `Composite` is `!Clone`, so the example is loaded twice.
    #[test]
    fn r_breakout_example_loaded_runs_identically_to_the_carved_signal() {
        let env = project::Env::std();
        let loaded = blueprint_from_json(include_str!("../examples/r_breakout.json"), &|t| std_vocabulary(t))
            .expect("shipped r_breakout example loads");
        let via_file = run_signal_r(loaded, &[], RunData::Synthetic, 0, &env);
        let via_carve = run_signal_r(r_breakout_signal(Some(R_BREAKOUT_CHANNEL)), &[], RunData::Synthetic, 0, &env);
        assert_eq!(via_file.metrics, via_carve.metrics, "loaded example grades identically to the carve");
    }

    /// Regenerates the shipped r_meanrev examples from the carved signal. Run by hand:
    /// `cargo test --bin aura emit_r_meanrev_examples -- --ignored`. The examples are
    /// green-by-construction (serialised from the builder, never hand-authored).
    #[test]
    #[ignore = "regenerates crates/aura-cli/examples/r_meanrev{,_open}.json; run by hand"]
    fn emit_r_meanrev_examples() {
        std::fs::create_dir_all("examples").expect("examples dir");
        std::fs::write(
            "examples/r_meanrev.json",
            blueprint_to_json(&r_meanrev_signal(Some(R_MEANREV_WINDOW), Some(R_MEANREV_BAND_K)))
                .expect("serialize closed r_meanrev"),
        )
        .expect("write examples/r_meanrev.json");
        std::fs::write(
            "tests/fixtures/r_meanrev_open.json",
            blueprint_to_json(&r_meanrev_signal(None, None)).expect("serialize open r_meanrev"),
        )
        .expect("write tests/fixtures/r_meanrev_open.json");
    }

    /// Regenerates the shipped r_channel examples from the carved signal. Run by hand:
    /// `cargo test --bin aura emit_r_channel_examples -- --ignored`. The examples are
    /// green-by-construction (serialised from the builder, never hand-authored).
    #[test]
    #[ignore = "regenerates crates/aura-cli/examples/r_channel{,_open}.json; run by hand"]
    fn emit_r_channel_examples() {
        std::fs::create_dir_all("examples").expect("examples dir");
        std::fs::write(
            "examples/r_channel.json",
            blueprint_to_json(&r_channel_signal(Some(R_CHANNEL_LENGTH))).expect("serialize closed r_channel"),
        )
        .expect("write examples/r_channel.json");
        std::fs::write(
            "tests/fixtures/r_channel_open.json",
            blueprint_to_json(&r_channel_signal(None)).expect("serialize open r_channel"),
        )
        .expect("write tests/fixtures/r_channel_open.json");
    }

    /// The shipped examples are a faithful serialisation of the carved signal
    /// (the r_breakout pin pattern): re-serialising the carve equals the
    /// checked-in bytes; a drift between builder and file breaks it.
    #[test]
    fn shipped_r_channel_examples_serialize_the_carved_signal() {
        assert_eq!(
            blueprint_to_json(&r_channel_signal(Some(R_CHANNEL_LENGTH))).expect("serialize closed"),
            include_str!("../examples/r_channel.json"),
        );
        assert_eq!(
            blueprint_to_json(&r_channel_signal(None)).expect("serialize open"),
            include_str!("../tests/fixtures/r_channel_open.json"),
        );
    }

    /// A 10-bar high/low/close fixture (canonical column order: high, low,
    /// close) that warms the hl_channel graph (Delay(1) + Rolling(3)) and
    /// breaks out upward then downward.
    fn hlc_sources() -> Vec<Box<dyn aura_engine::Source>> {
        let high = [10.5_f64, 10.6, 10.4, 10.8, 11.5, 12.0, 12.2, 11.0, 10.2, 9.8];
        let low = [9.5_f64, 9.7, 9.6, 9.9, 10.8, 11.4, 11.6, 10.1, 9.4, 9.0];
        let close = [10.0_f64, 10.2, 10.0, 10.5, 11.3, 11.8, 12.0, 10.4, 9.6, 9.2];
        [&high[..], &low[..], &close[..]]
            .into_iter()
            .map(|col| {
                let series: Vec<(Timestamp, Scalar)> = col
                    .iter()
                    .enumerate()
                    .map(|(i, &v)| (Timestamp(i as i64 + 1), Scalar::f64(v)))
                    .collect();
                Box::new(VecSource::new(series)) as Box<dyn aura_engine::Source>
            })
            .collect()
    }

    /// The shipped closed example, reloaded through the data plane and run
    /// through the MULTI-COLUMN wrap, emits bit-identical bias rows to the
    /// carved signal — r_channel's durable equivalence anchor (the r_breakout
    /// idiom, over three VecSource columns instead of RunData::Synthetic,
    /// which honestly refuses multi-column signals). `wrap_r`'s ex tap reads
    /// `sig.output("bias")` directly, so `rx_ex` carries the raw signal bias.
    #[test]
    fn r_channel_example_loaded_runs_identically_to_the_carved_signal() {
        let run = |signal: Composite| -> Vec<(Timestamp, Vec<Scalar>)> {
            let binding =
                binding::resolve_binding(signal.name(), signal.input_roles(), &BTreeMap::new())
                    .expect("high/low/close roles resolve");
            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 = wrap_r(
                signal,
                tx_eq,
                tx_ex,
                tx_r,
                tx_req,
                StopRule::Vol { length: 3, k: 2.0 },
                false,
                SYNTHETIC_PIP_SIZE,
                &binding,
                None,
            )
            .compile_with_params(&[])
            .expect("closed hl_channel wraps to a valid harness");
            let mut h = Harness::bootstrap(flat).expect("hl_channel harness bootstraps");
            h.run(hlc_sources());
            rx_ex.try_iter().collect()
        };
        let loaded = blueprint_from_json(include_str!("../examples/r_channel.json"), &|t| std_vocabulary(t))
            .expect("shipped r_channel example loads");
        let via_file = run(loaded);
        let via_carve = run(r_channel_signal(Some(R_CHANNEL_LENGTH)));
        assert!(!via_carve.is_empty(), "the channel must emit bias rows over the fixture");
        assert!(
            via_carve.iter().any(|(_, row)| row.iter().any(|s| s.as_f64() != 0.0)),
            "the fixture must drive a non-zero bias (an all-zero run would make \
             the equivalence vacuous)"
        );
        assert_eq!(via_file, via_carve, "loaded example emits the carve's bias rows");
    }

    /// The shipped examples are a faithful serialisation of the carved signal (#159 cut 3):
    /// re-serialising the carve equals the checked-in bytes. Green-by-construction with the
    /// emitter; a drift between builder and file breaks it.
    #[test]
    fn shipped_r_meanrev_examples_serialize_the_carved_signal() {
        assert_eq!(
            blueprint_to_json(&r_meanrev_signal(Some(R_MEANREV_WINDOW), Some(R_MEANREV_BAND_K))).expect("closed"),
            include_str!("../examples/r_meanrev.json"),
        );
        assert_eq!(
            blueprint_to_json(&r_meanrev_signal(None, None)).expect("open"),
            include_str!("../tests/fixtures/r_meanrev_open.json"),
        );
    }

    /// The shipped closed example, reloaded through the data plane and run, grades
    /// bit-identically to running the carved signal directly (#159 cut 3). This is
    /// r_meanrev's durable equivalence anchor after the fused builder retires: it pins
    /// that `examples/r_meanrev.json` still produces the carve's grade, with no
    /// hardcoded golden. `Composite` is `!Clone`, so the example is loaded twice.
    #[test]
    fn r_meanrev_example_loaded_runs_identically_to_the_carved_signal() {
        let env = project::Env::std();
        let loaded = blueprint_from_json(include_str!("../examples/r_meanrev.json"), &|t| std_vocabulary(t))
            .expect("shipped r_meanrev example loads");
        let via_file = run_signal_r(loaded, &[], RunData::Synthetic, 0, &env);
        let via_carve = run_signal_r(
            r_meanrev_signal(Some(R_MEANREV_WINDOW), Some(R_MEANREV_BAND_K)),
            &[],
            RunData::Synthetic,
            0,
            &env,
        );
        assert_eq!(via_file.metrics, via_carve.metrics, "loaded example grades identically to the carve");
    }

    /// Independently pins the shipped `r_meanrev_signal` carve's FADE direction —
    /// short (-1) above the band, long (+1) below — using the Scale-based band it
    /// actually ships with. The equivalence anchor above only proves the loaded
    /// example agrees with the carve; both sides run the same `r_meanrev_signal`, so
    /// it is tautological on carve correctness (a mis-wired or sign-inverted carve
    /// would fail identically on both sides and still pass). The retired
    /// `r_meanrev_graph` builder (LinComb-based) that once graded against this exact
    /// polarity is deleted (#159 cut 3); `aura-engine`'s `r_meanrev_e2e` survives but
    /// still hand-builds the band with `LinComb(1)`, not `Scale`, so it does not
    /// exercise the node this carve actually ships. `k = 0` collapses the band to the
    /// lagging EWMA mean, isolating direction + latch from the sigma threshold (same
    /// idiom as `r_meanrev_e2e`'s own k=0 case); window 3 (alpha = 0.5) lags the level
    /// clearly. `wrap_r`'s `ex` tap reads `sig.output("bias")` directly (before any
    /// broker/exec/cost machinery), so `rx_ex` carries the raw signal bias.
    #[test]
    fn r_meanrev_signal_fades_short_above_the_band_and_long_below() {
        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 signal = r_meanrev_signal(Some(3), Some(0.0));
        let binding = binding::resolve_binding(signal.name(), signal.input_roles(), &BTreeMap::new())
            .expect("the price role resolves");
        let flat = wrap_r(
            signal,
            tx_eq,
            tx_ex,
            tx_r,
            tx_req,
            StopRule::Vol { length: 3, k: 2.0 },
            false,
            SYNTHETIC_PIP_SIZE,
            &binding,
            None,
        )
        .compile_with_params(&[])
        .expect("r-meanrev signal wraps to a valid harness");
        let mut h = Harness::bootstrap(flat).expect("r-meanrev harness bootstraps");
        // calm (price == lagging mean -> no fade) | sustained UP (price > mean ->
        // fade SHORT) | sustained DOWN (price < mean -> fade LONG).
        let closes = [
            100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 130.0, 130.0, 130.0, 70.0, 70.0, 70.0, 70.0,
        ];
        let prices: Vec<(Timestamp, Scalar)> =
            closes.iter().enumerate().map(|(i, &c)| (Timestamp(i as i64), Scalar::f64(c))).collect();
        let src: Vec<Box<dyn aura_engine::Source>> = vec![Box::new(VecSource::new(prices))];
        h.run(src);
        let bias: Vec<f64> =
            rx_ex.try_iter().map(|(_, row): (Timestamp, Vec<Scalar>)| row[0].as_f64()).collect();
        assert!(!bias.is_empty(), "the meanrev exposure tap must emit once warmed up");
        assert_eq!(*bias.first().unwrap(), 0.0, "calm bars (price == mean) must not fade: {bias:?}");
        let first_short = bias.iter().position(|&b| b == -1.0).expect("an up-move must fade SHORT (-1)");
        let first_long = bias.iter().position(|&b| b == 1.0).expect("a down-move must fade LONG (+1)");
        assert!(first_short < first_long, "short (up-fade) must precede long (down-fade): {bias:?}");
        assert_eq!(*bias.last().unwrap(), 1.0, "the down-fade long must hold to the end: {bias:?}");
    }

    /// #234: the optional cost leg records an aggregate cost stream positionally
    /// 1:1 with the R record (`summarize_r`'s positional-join contract) plus a
    /// per-cycle net_r_equity curve, in non-reduce trace mode. The cost-less
    /// side (`cost: None` == today's graph, byte-identical) is pinned by the
    /// suite's untouched equivalence anchors.
    #[test]
    fn wrap_r_cost_leg_records_cost_rows_co_temporal_with_the_r_record() {
        let signal = load_closed_r_sma();
        let binding = binding::resolve_binding(signal.name(), signal.input_roles(), &BTreeMap::new())
            .expect("the price role resolves");
        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 (tx_cost, rx_cost) = mpsc::channel();
        let (tx_net, rx_net) = mpsc::channel();
        let leg = CostLeg {
            nodes: vec![ConstantCost::builder().bind("cost_per_trade", Scalar::f64(0.0005))],
            tx_cost,
            tx_net,
        };
        let flat = wrap_r(
            signal,
            tx_eq,
            tx_ex,
            tx_r,
            tx_req,
            StopRule::Vol { length: R_SMA_STOP_LENGTH, k: R_SMA_STOP_K },
            false,
            SYNTHETIC_PIP_SIZE,
            &binding,
            Some(leg),
        )
        .compile_with_params(&[])
        .expect("costed wrap builds");
        let mut h = Harness::bootstrap(flat).expect("costed harness bootstraps");
        let src: Vec<Box<dyn aura_engine::Source>> = vec![Box::new(VecSource::new(r_sma_prices()))];
        h.run(src);
        let r_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_r.try_iter().collect();
        let cost_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_cost.try_iter().collect();
        let net_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_net.try_iter().collect();
        assert!(!r_rows.is_empty(), "the fixture must warm the executor");
        assert_eq!(
            cost_rows.len(),
            r_rows.len(),
            "the cost stream is positionally 1:1 with the R record (co-temporality)"
        );
        assert_eq!(net_rows.len(), cost_rows.len(), "the net curve emits per cost row");
        assert!(
            cost_rows.iter().any(|(_, row)| row[0].as_f64() != 0.0),
            "at least one close must charge a non-zero cost (non-vacuous)"
        );
    }

    /// #234: the reduce-mode cost branch (a `GatedRecorder` with the appended
    /// `closed_this_cycle` gate column) stays co-temporal with the gated R
    /// record too — the same 1:1 join contract as the trace-mode leg above,
    /// but over the member/sweep run path `summarize_r` actually consumes.
    #[test]
    fn wrap_r_cost_leg_in_reduce_mode_stays_co_temporal_with_the_gated_r_record() {
        let signal = load_closed_r_sma();
        let binding = binding::resolve_binding(signal.name(), signal.input_roles(), &BTreeMap::new())
            .expect("the price role resolves");
        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 (tx_cost, rx_cost) = mpsc::channel();
        let (tx_net, _rx_net) = mpsc::channel();
        let leg = CostLeg {
            nodes: vec![ConstantCost::builder().bind("cost_per_trade", Scalar::f64(0.0005))],
            tx_cost,
            tx_net,
        };
        let flat = wrap_r(
            signal,
            tx_eq,
            tx_ex,
            tx_r,
            tx_req,
            StopRule::Vol { length: R_SMA_STOP_LENGTH, k: R_SMA_STOP_K },
            true,
            SYNTHETIC_PIP_SIZE,
            &binding,
            Some(leg),
        )
        .compile_with_params(&[])
        .expect("costed reduce-mode wrap builds");
        let mut h = Harness::bootstrap(flat).expect("costed reduce-mode harness bootstraps");
        let src: Vec<Box<dyn aura_engine::Source>> = vec![Box::new(VecSource::new(r_sma_prices()))];
        h.run(src);
        let r_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_r.try_iter().collect();
        let cost_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_cost.try_iter().collect();
        assert!(!r_rows.is_empty(), "the fixture must warm and close at least one trade");
        assert_eq!(
            cost_rows.len(),
            r_rows.len(),
            "the reduce-mode gated cost stream is positionally 1:1 with the gated R record"
        );
        assert!(
            cost_rows.iter().any(|(_, row)| row[0].as_f64() != 0.0),
            "at least one gated close must charge a non-zero cost (non-vacuous)"
        );
    }

    /// #234: a cost component that declares an extra `volatility` port (the
    /// `VolSlippageCost` shape) is wired to a live realized-range proxy built
    /// from the close role (`RollingMax`/`RollingMin`/`Sub` over
    /// `SLIP_VOL_LENGTH`), discovered purely from the component's own schema
    /// past the geometry prefix (`vol_slots`) — never a disconnected or
    /// hardwired-zero input. If that wiring regresses, `ctx.f64_in(GEOMETRY_WIDTH)`
    /// inside the component stays permanently empty and every charge is exactly
    /// 0.0 (`vol_not_yet_warm_emits_zero_cost_co_temporally`'s withhold case), so a
    /// non-zero charge over a genuinely moving price series is a direct witness
    /// that the proxy reached the component. The two tests above use
    /// `ConstantCost`, which never touches this wiring at all.
    #[test]
    fn wrap_r_cost_leg_wires_the_vol_slippage_proxy_from_the_close_role() {
        let signal = load_closed_r_sma();
        let binding = binding::resolve_binding(signal.name(), signal.input_roles(), &BTreeMap::new())
            .expect("the price role resolves");
        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 (tx_cost, rx_cost) = mpsc::channel();
        let (tx_net, _rx_net) = mpsc::channel();
        let leg = CostLeg {
            nodes: vec![VolSlippageCost::builder().bind("slip_vol_mult", Scalar::f64(0.5))],
            tx_cost,
            tx_net,
        };
        let flat = wrap_r(
            signal,
            tx_eq,
            tx_ex,
            tx_r,
            tx_req,
            StopRule::Vol { length: R_SMA_STOP_LENGTH, k: R_SMA_STOP_K },
            false,
            SYNTHETIC_PIP_SIZE,
            &binding,
            Some(leg),
        )
        .compile_with_params(&[])
        .expect("vol-slippage-costed wrap builds");
        let mut h = Harness::bootstrap(flat).expect("vol-slippage-costed harness bootstraps");
        let src: Vec<Box<dyn aura_engine::Source>> = vec![Box::new(VecSource::new(r_sma_prices()))];
        h.run(src);
        let r_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_r.try_iter().collect();
        let cost_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_cost.try_iter().collect();
        assert!(!r_rows.is_empty(), "the fixture must warm and close at least one trade");
        assert_eq!(
            cost_rows.len(),
            r_rows.len(),
            "the vol-slippage cost stream stays positionally 1:1 with the R record"
        );
        assert!(
            cost_rows.iter().any(|(_, row)| row[0].as_f64() != 0.0),
            "the volatility proxy must actually feed the component — a disconnected \
             proxy leaves cost_in_r at exactly 0.0 on every close: {cost_rows:?}"
        );
    }

    /// #234: a member run under a constant cost model nets the hand-computed
    /// per-trade cost — `net_expectancy_r ≈ expectancy_r − mean(cost_per_trade
    /// / |entry − stop|)` over summarize_r's ledger (closed rows + the
    /// window-end open row; the CostRunner R-normalization contract), while
    /// every gross field stays byte-identical to the cost-less run; and the
    /// manifest stamps the component for reproduce to re-derive.
    #[test]
    fn run_blueprint_member_joins_a_constant_cost_model_into_net_metrics() {
        let env = project::Env::std();
        let data = DataSource::Synthetic;
        let doc = blueprint_to_json(&load_closed_r_sma()).expect("serializes");
        let reload = || blueprint_from_json(&doc, &|t| std_vocabulary(t)).expect("loads");
        let space = blueprint_axis_probe(&doc, &env).param_space();
        let binding = binding::resolve_binding("costnet", reload().input_roles(), &BTreeMap::new())
            .expect("the price role resolves");
        let stop = StopRule::Vol { length: R_SMA_STOP_LENGTH, k: R_SMA_STOP_K };
        let window = data.full_window(&env);
        let pip = data.pip_size();
        const CPT: f64 = 0.0005; // price units; the synthetic stream trades near 1.0
        let run = |cost: &[aura_research::CostSpec]| {
            run_blueprint_member(
                reload(),
                &[],
                &space,
                data.run_sources(&env, &binding.columns()),
                window,
                0,
                pip,
                "topo",
                &env,
                stop,
                &binding,
                cost,
                "GER40",
            )
        };
        let gross = run(&[]);
        let netted = run(&[aura_research::CostSpec::Constant {
            cost_per_trade: aura_research::CostValue::Scalar(CPT),
        }]);

        // The trade geometry, independently: a non-reduce cost-less wrap over
        // the same realization, draining the dense R record whose ledger
        // summarize_r reads.
        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 = wrap_r(reload(), tx_eq, tx_ex, tx_r, tx_req, stop, false, pip, &binding, None)
            .compile_with_params(&[])
            .expect("wraps");
        let mut h = Harness::bootstrap(flat).expect("bootstraps");
        h.run(data.run_sources(&env, &binding.columns()));
        let r_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_r.try_iter().collect();

        // summarize_r's ledger: closed rows, plus the final row if open — each
        // charged cost_per_trade / |entry − stop| (0 when the latched distance
        // is 0). PM record cols: closed=0, entry_price=6, stop_price=7,
        // open=11 (the aura-analysis r_col contract).
        let mut costs: Vec<f64> = Vec::new();
        for (i, (_, row)) in r_rows.iter().enumerate() {
            let is_last = i == r_rows.len() - 1;
            if row[0].as_bool() || (is_last && row[11].as_bool()) {
                let latched = (row[6].as_f64() - row[7].as_f64()).abs();
                costs.push(if latched > 0.0 { CPT / latched } else { 0.0 });
            }
        }
        let g = gross.metrics.r.as_ref().expect("gross member carries R metrics");
        let n = netted.metrics.r.as_ref().expect("netted member carries R metrics");
        assert_eq!(
            costs.len() as u64, g.n_trades,
            "the independent ledger walk must see the member's trades"
        );
        assert!(costs.iter().any(|&c| c > 0.0), "at least one trade must charge (non-vacuous)");
        let mean_cost = costs.iter().sum::<f64>() / costs.len() as f64;

        assert_eq!(gross.metrics.total_pips, netted.metrics.total_pips, "gross pips unchanged");
        assert_eq!(g.expectancy_r, n.expectancy_r, "gross R stays byte-identical under cost");
        assert_eq!(g.n_trades, n.n_trades);
        assert_eq!(g.net_expectancy_r, g.expectancy_r, "cost-less: net == gross");
        assert!(
            (n.net_expectancy_r - (g.expectancy_r - mean_cost)).abs() < 1e-12,
            "net = gross − mean per-trade cost: net {} gross {} mean_cost {mean_cost}",
            n.net_expectancy_r,
            g.expectancy_r
        );
        // The manifest stamps the component (the reproduce re-derivation carrier)...
        assert!(
            netted
                .manifest
                .params
                .iter()
                .any(|(k, v)| k == "cost[0].cost_per_trade" && *v == Scalar::f64(CPT)),
            "member manifest stamps the cost component: {:?}",
            netted.manifest.params
        );
        // ...and a cost-less member stamps nothing (content/label stability).
        assert!(
            !gross.manifest.params.iter().any(|(k, _)| k.starts_with("cost[")),
            "a cost-less member stamps no cost params"
        );
    }

    /// #234: `campaign_run::persist_campaign_traces`'s C1 drift alarm re-runs a
    /// costed member in `!reduce` mode (fresh per-cycle taps + a `CostLeg`
    /// bound the same way `cost_nodes_for` binds it) and compares the result
    /// against the recorded reduce-mode member. This pins that cross-mode
    /// equivalence holds UNDER a non-empty cost model: the `!reduce` +
    /// `CostLeg` wiring (this test, verbatim) and the reduce-mode
    /// `run_blueprint_member` path must net to the exact same `RunMetrics`
    /// over the same synthetic realization — the equality the drift alarm
    /// structurally depends on to not false-positive on every legitimate
    /// costed campaign.
    #[test]
    fn persist_side_nonreduce_rerun_matches_reduce_mode_net_metrics_under_cost() {
        let env = project::Env::std();
        let data = DataSource::Synthetic;
        let doc = blueprint_to_json(&load_closed_r_sma()).expect("serializes");
        let reload = || blueprint_from_json(&doc, &|t| std_vocabulary(t)).expect("loads");
        let space = blueprint_axis_probe(&doc, &env).param_space();
        let binding =
            binding::resolve_binding("persistnet", reload().input_roles(), &BTreeMap::new())
                .expect("the price role resolves");
        let stop = StopRule::Vol { length: R_SMA_STOP_LENGTH, k: R_SMA_STOP_K };
        let window = data.full_window(&env);
        let pip = data.pip_size();
        let cost = [aura_research::CostSpec::Constant {
            cost_per_trade: aura_research::CostValue::Scalar(0.0005),
        }];

        // The recorded member: reduce-mode, cost-bound — `run_blueprint_member`,
        // the exact path `CliMemberRunner::run_member` calls.
        let recorded = run_blueprint_member(
            reload(),
            &[],
            &space,
            data.run_sources(&env, &binding.columns()),
            window,
            0,
            pip,
            "topo",
            &env,
            stop,
            &binding,
            &cost,
            "GER40",
        );

        // The persist-side re-run, verbatim structure: `!reduce` + a `CostLeg`
        // built through the SAME `campaign_run::cost_nodes_for`, then
        // `summarize` + `summarize_r(&r_rows, &cost_rows)` — exactly
        // `persist_campaign_traces`'s C1 drift-alarm computation.
        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 (tx_cost, rx_cost) = mpsc::channel();
        let (tx_net, _rx_net) = mpsc::channel();
        let cost_leg =
            Some(CostLeg { nodes: campaign_run::cost_nodes_for(&cost, "GER40"), tx_cost, tx_net });
        let flat = wrap_r(reload(), tx_eq, tx_ex, tx_r, tx_req, stop, false, pip, &binding, cost_leg)
            .compile_with_params(&[])
            .expect("the persist-side wrap builds");
        let mut h = Harness::bootstrap(flat).expect("the persist-side harness bootstraps");
        h.run(data.run_sources(&env, &binding.columns()));
        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 cost_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_cost.try_iter().collect();
        let mut rerun_metrics = summarize(&f64_field(&eq_rows, 0), &f64_field(&ex_rows, 0));
        rerun_metrics.r = Some(summarize_r(&r_rows, &cost_rows));

        assert!(
            rerun_metrics.r.as_ref().is_some_and(|r| r.n_trades > 0),
            "the fixture must actually close at least one costed trade"
        );
        assert_eq!(
            rerun_metrics, recorded.metrics,
            "the persist path's !reduce + CostLeg re-run must net to the exact same \
             RunMetrics as the reduce-mode member under the same cost model — the \
             C1 drift alarm `persist_campaign_traces` relies on"
        );
    }

    /// Property: the pip a run resolves and stamps into its manifest must be the
    /// pip the in-graph `SimBroker` divides by — the resolved (real, per-instrument)
    /// pip has to reach the graph, not just the manifest label. The `SimBroker`
    /// integrates `exposure * (price - prev_price) / pip`, so the raw price-move
    /// total `total_pips * pip` is pip-invariant: the SAME signal + prices run at
    /// two different pips must agree on that product. If `pip` only decorates the
    /// label and the graph always divides by the synthetic default, both runs
    /// yield the identical `total_pips` and the invariant breaks (and real-data
    /// pips are inflated by `1 / 0.0001 = 10^4`).
    #[test]
    fn run_blueprint_member_computes_pips_at_the_resolved_pip_not_a_hardwired_default() {
        let env = project::Env::std();
        // A price series that drives the meanrev signal into a definite non-flat
        // exposure (short an up-move, long a down-move), so broker equity != 0 —
        // the same idiom as the fade test above.
        let closes = [
            100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 130.0, 130.0, 130.0, 70.0, 70.0, 70.0, 70.0,
        ];
        let make_source = || -> Vec<Box<dyn aura_engine::Source>> {
            let prices: Vec<(Timestamp, Scalar)> = closes
                .iter()
                .enumerate()
                .map(|(i, &c)| (Timestamp(i as i64), Scalar::f64(c)))
                .collect();
            vec![Box::new(VecSource::new(prices))]
        };
        let window = (Timestamp(0), Timestamp(closes.len() as i64 - 1));
        let stop = StopRule::Vol { length: 3, k: 2.0 };
        let space: Vec<ParamSpec> = vec![];
        // Two per-instrument pips, an order of magnitude apart and both distinct
        // from the synthetic 0.0001 — i.e. the real-data condition.
        let pip_a = 1.0_f64;
        let pip_b = 0.1_f64;
        let signal_a = r_meanrev_signal(Some(3), Some(0.0));
        let binding = binding::resolve_binding(signal_a.name(), signal_a.input_roles(), &BTreeMap::new())
            .expect("the price role resolves");
        let report_a = run_blueprint_member(
            signal_a, &[], &space, make_source(), window, 0, pip_a, "topo", &env, stop, &binding, &[], "GER40",
        );
        let report_b = run_blueprint_member(
            r_meanrev_signal(Some(3), Some(0.0)), &[], &space, make_source(), window, 0, pip_b, "topo", &env, stop, &binding, &[], "GER40",
        );
        // Guard: the run must have actually traded, else the invariant is vacuous.
        assert!(
            report_a.metrics.total_pips.abs() > 0.0,
            "test needs a non-flat run to be meaningful: {:?}",
            report_a.metrics
        );
        // The pip-invariant raw price-move total must agree across the two pips.
        let raw_a = report_a.metrics.total_pips * pip_a;
        let raw_b = report_b.metrics.total_pips * pip_b;
        assert!(
            (raw_a - raw_b).abs() < 1e-9,
            "total_pips must be computed at the resolved pip: pip={pip_a} gave total_pips={} \
             (raw price-move {raw_a}), pip={pip_b} gave total_pips={} (raw price-move {raw_b}) \
             — the resolved pip never reached the in-graph SimBroker",
            report_a.metrics.total_pips,
            report_b.metrics.total_pips,
        );
    }

    /// Loads the shipped closed r-sma example (fast=2, slow=4 bound) through the
    /// public `blueprint_from_json` path — the single call site so a fixture
    /// rename or vocabulary change is one edit, not fourteen.
    fn load_closed_r_sma() -> Composite {
        blueprint_from_json(include_str!("../examples/r_sma.json"), &|t| std_vocabulary(t))
            .expect("loads")
    }

    /// Loads the shipped open r-sma example (both SMA lengths free) through the
    /// public `blueprint_from_json` path.
    fn load_open_r_sma() -> Composite {
        blueprint_from_json(include_str!("../tests/fixtures/r_sma_open.json"), &|t| std_vocabulary(t))
            .expect("loads")
    }

    #[test]
    fn select_winner_refuses_plateau_without_a_lattice() {
        // A plateau request with no lattice (a random sweep would yield None) is
        // refused, never silently argmaxed. The refuse short-circuits before the
        // family is read, so an empty family is fine here.
        let fam = SweepFamily { space: vec![], points: vec![] };
        let err = select_winner(&fam, "total_pips", Selection::Plateau(PlateauMode::Mean), None)
            .unwrap_err();
        assert!(err.contains("requires a grid sweep"), "refuse message: {err}");
    }

    /// The `--select` token grammar (`parse_select`) maps argmax / plateau:mean /
    /// plateau:worst and rejects an unknown token — the pure selector clap's
    /// `--select` value feeds on both the built-in and blueprint walk-forward paths.
    #[test]
    fn parse_select_token_grammar() {
        assert!(matches!(parse_select("argmax").unwrap(), Selection::Argmax), "default is argmax");
        assert!(matches!(parse_select("plateau:mean").unwrap(), Selection::Plateau(PlateauMode::Mean)));
        assert!(matches!(parse_select("plateau:worst").unwrap(), Selection::Plateau(PlateauMode::Worst)));
        assert!(parse_select("bogus").is_err(), "unknown --select token is a usage error");
    }

    fn cmp_member(key: &str, ts: &[i64], vals: &[f64]) -> FamilyMember {
        cmp_member_win(key, ts, vals, (0, 0))
    }

    /// Like [`cmp_member`] but with an explicit manifest `window` so a test can
    /// model walk-forward members (disjoint per-member OOS windows) and assert the
    /// family window spans them.
    fn cmp_member_win(key: &str, ts: &[i64], vals: &[f64], window: (i64, i64)) -> FamilyMember {
        let rows: Vec<(Timestamp, Vec<Scalar>)> =
            ts.iter().zip(vals).map(|(&t, &v)| (Timestamp(t), vec![Scalar::f64(v)])).collect();
        let tap = ColumnarTrace::from_rows("equity", &[ScalarKind::F64], &rows);
        FamilyMember {
            key: key.to_string(),
            traces: RunTraces {
                manifest: sim_optimal_manifest(
                    vec![],
                    (Timestamp(window.0), Timestamp(window.1)),
                    0,
                    1.0,
                ),
                taps: vec![tap],
            },
        }
    }

    #[test]
    fn comparison_overlays_one_shared_scale_series_per_member() {
        let members = vec![
            cmp_member("a", &[1, 2, 3], &[10.0, 11.0, 12.0]),
            cmp_member("b", &[1, 2, 3], &[20.0, 21.0, 22.0]),
        ];
        let data = build_comparison_chart_data("fam", &members, "equity").expect("builds");
        assert_eq!(data.xs, vec![1, 2, 3]);
        assert_eq!(data.series.len(), 2);
        assert_eq!(data.series[0].name, "a");
        assert_eq!(data.series[1].name, "b");
        // ONE shared y-scale across members (same quantity).
        assert_eq!(data.series[0].y_scale_id, data.series[1].y_scale_id);
        // shared ts -> dense, no nulls.
        assert!(data.series[0].points.iter().all(Option::is_some));
        // #102 meta wiring: a family carries kind/name/member-count + the one
        // compared tap, and never the per-member params (those are the labels).
        assert_eq!(data.meta.kind, "family");
        assert_eq!(data.meta.name, "fam");
        assert_eq!(data.meta.members, Some(2));
        assert_eq!(data.meta.taps, vec!["equity".to_string()]);
        assert!(data.meta.params.is_empty(), "family meta must not repeat per-member params");
    }

    #[test]
    fn comparison_disjoint_members_are_null_complementary() {
        let members = vec![
            cmp_member("oos1", &[1, 2], &[10.0, 11.0]),
            cmp_member("oos2", &[3, 4], &[20.0, 21.0]),
        ];
        let data = build_comparison_chart_data("fam", &members, "equity").expect("builds");
        assert_eq!(data.xs, vec![1, 2, 3, 4]);
        assert_eq!(data.series[0].points, vec![Some(10.0), Some(11.0), None, None]);
        assert_eq!(data.series[1].points, vec![None, None, Some(20.0), Some(21.0)]);
    }

    /// #102 family-window semantics: the header's `window` for a family is the
    /// SPAN across all members — `(min member.from, max member.to)` — not the first
    /// member's window. The distinction is load-bearing for a walk-forward family,
    /// whose members are DISJOINT OOS windows (commit 4c64feb): labelling such a
    /// family with `members[0]`'s window mislabels the family's true coverage. The
    /// span reading is correct for all three kinds (sweep/MC members share a window,
    /// so their span collapses to that shared window).
    #[test]
    fn comparison_window_spans_disjoint_walk_forward_members() {
        let members = vec![
            cmp_member_win("oos1", &[10, 20], &[1.0, 2.0], (10, 20)),
            cmp_member_win("oos2", &[30, 40], &[3.0, 4.0], (30, 40)),
            cmp_member_win("oos3", &[50, 60], &[5.0, 6.0], (50, 60)),
        ];
        let data = build_comparison_chart_data("wf", &members, "equity").expect("builds");
        // SPAN of all OOS windows (10..60), NOT members[0]'s window (10..20).
        assert_eq!(data.meta.window, (10, 60));
    }

    #[test]
    fn comparison_errors_when_no_member_has_the_tap() {
        let members = vec![cmp_member("a", &[1], &[1.0])];
        assert!(build_comparison_chart_data("fam", &members, "nosuch").is_err());
    }

    #[test]
    fn decimate_bounds_the_spine_to_twice_the_bucket_count() {
        let n = 10_000usize;
        let xs: Vec<i64> = (0..n as i64).collect();
        let points: Vec<Option<f64>> = (0..n).map(|i| Some(i as f64)).collect();
        let data = ChartData {
            xs,
            series: vec![Series { name: "equity".into(), y_scale_id: "y_0".into(), points, reduce: ReduceKind::MinMax }],
            meta: ChartMeta::default(),
        };
        let out = decimate(data, 2000);
        assert!(out.xs.len() <= 4000, "spine not bounded: {}", out.xs.len());
        assert_eq!(out.xs.len(), out.series[0].points.len(), "xs and points must stay aligned");
    }

    #[test]
    fn decimate_preserves_per_bucket_min_and_max() {
        // 10 points, 2 buckets -> bucket 0 = idx 0..5 (a spike), bucket 1 = idx 5..10 (a trough).
        let xs: Vec<i64> = (0..10).collect();
        let mut pv = vec![1.0_f64; 10];
        pv[3] = 999.0;
        pv[7] = -50.0;
        let points: Vec<Option<f64>> = pv.into_iter().map(Some).collect();
        let data = ChartData {
            xs,
            series: vec![Series { name: "equity".into(), y_scale_id: "y_0".into(), points, reduce: ReduceKind::MinMax }],
            meta: ChartMeta::default(),
        };
        let out = decimate(data, 2);
        let got = out.series[0].points.clone();
        assert!(got.contains(&Some(999.0)), "bucket max (spike) dropped: {got:?}");
        assert!(got.contains(&Some(-50.0)), "bucket min (trough) dropped: {got:?}");
    }

    #[test]
    fn decimate_keeps_an_all_null_bucket_null() {
        let xs: Vec<i64> = (0..10).collect();
        let mut points: Vec<Option<f64>> = (0..5).map(|i| Some(i as f64)).collect();
        points.extend(std::iter::repeat_n(None, 5));
        let data = ChartData {
            xs,
            series: vec![Series { name: "equity".into(), y_scale_id: "y_0".into(), points, reduce: ReduceKind::MinMax }],
            meta: ChartMeta::default(),
        };
        let out = decimate(data, 2);
        assert_eq!(*out.series[0].points.last().unwrap(), None, "all-null bucket must stay null");
    }

    #[test]
    fn decimate_is_a_noop_within_budget() {
        let data = ChartData {
            xs: vec![1, 2, 3],
            series: vec![Series { name: "equity".into(), y_scale_id: "y_0".into(), points: vec![Some(1.0), Some(2.0), Some(3.0)], reduce: ReduceKind::MinMax }],
            meta: ChartMeta::default(),
        };
        let out = decimate(data, 2000);
        assert_eq!(out.xs, vec![1, 2, 3], "within-budget data must pass through unchanged");
        assert_eq!(out.series[0].points, vec![Some(1.0), Some(2.0), Some(3.0)]);
    }

    #[test]
    fn decimate_passes_meta_through_and_keeps_xs_monotonic() {
        let n = 10_000usize;
        let xs: Vec<i64> = (0..n as i64).collect();
        let points: Vec<Option<f64>> = (0..n).map(|i| Some(i as f64)).collect();
        let meta = ChartMeta { name: "keep-me".into(), ..Default::default() };
        let data = ChartData { xs, series: vec![Series { name: "equity".into(), y_scale_id: "y_0".into(), points, reduce: ReduceKind::MinMax }], meta };
        let out = decimate(data, 2000);
        assert_eq!(out.meta.name, "keep-me", "meta must pass through decimation");
        assert!(out.xs.windows(2).all(|w| w[0] < w[1]), "decimated spine must stay strictly increasing");
    }

    /// #111: a bounded *level* series with `reduce = Mean` decimates to each bucket's
    /// MEAN, not its min/max envelope — so a high-flip bipolar exposure shows its
    /// net/duty-cycle level instead of collapsing to a -1..+1 band. RED under the
    /// shipped min/max-only decimation (any bucket holding a +1 emits +1); GREEN once
    /// `decimate` honours `ReduceKind::Mean`.
    #[test]
    fn decimate_mean_reduces_a_bipolar_series_to_its_bucket_level() {
        // 10 points, 2 buckets. Bucket 0 (idx 0..5) = [+1,+1,-1,+1,+1] -> mean +0.6;
        // bucket 1 (idx 5..10) = all -1 -> mean -1.0.
        let xs: Vec<i64> = (0..10).collect();
        let pv = vec![1.0, 1.0, -1.0, 1.0, 1.0, -1.0, -1.0, -1.0, -1.0, -1.0];
        let points: Vec<Option<f64>> = pv.into_iter().map(Some).collect();
        let data = ChartData {
            xs,
            series: vec![Series { name: "exposure".into(), y_scale_id: "y_0".into(), points, reduce: ReduceKind::Mean }],
            meta: ChartMeta::default(),
        };
        let out = decimate(data, 2);
        let got = out.series[0].points.clone();
        // No -1..+1 envelope: bucket 0 is its mean (+0.6), not a min/max pair.
        assert!(!got.contains(&Some(1.0)), "mean reduce must not emit a +1 envelope point: {got:?}");
        assert!(got.contains(&Some(0.6)), "bucket-0 duty-cycle mean (+0.6) missing: {got:?}");
        // bucket 0 spans two slots, both = the mean (a flat step, not a -1->+1 ramp).
        assert_eq!(got[0], Some(0.6), "first slot must be the bucket mean");
        assert_eq!(got[1], Some(0.6), "second slot must also be the bucket mean");
    }

    /// #102 single-run meta wiring: `build_chart_data` maps the `RunManifest` into
    /// `ChartData.meta` — kind "run", the name arg, the manifest window/broker, the
    /// charted taps, and the bound params stringified (each typed `Scalar` rendered
    /// via `render_value`, preserving its lexical form: `i64` decimal, `f64`
    /// shortest round-trip). A single run carries no member count.
    #[test]
    fn build_chart_data_threads_run_manifest_into_meta() {
        let eq_rows: Vec<(Timestamp, Vec<Scalar>)> =
            [1i64, 2, 3].iter().map(|&t| (Timestamp(t), vec![Scalar::f64(t as f64)])).collect();
        let traces = RunTraces {
            manifest: sim_optimal_manifest(
                vec![("len".into(), Scalar::i64(10)), ("scale".into(), Scalar::f64(0.5))],
                (Timestamp(1), Timestamp(3)),
                7,
                1.0,
            ),
            taps: vec![ColumnarTrace::from_rows("equity", &[ScalarKind::F64], &eq_rows)],
        };
        let data = build_chart_data("demo", traces);
        let meta = &data.meta;
        assert_eq!(meta.kind, "run");
        assert_eq!(meta.name, "demo");
        assert_eq!(meta.window, (1, 3));
        assert_eq!(meta.broker, "sim-optimal(pip_size=1)");
        assert_eq!(meta.seed, 7);
        assert_eq!(meta.taps, vec!["equity".to_string()]);
        assert_eq!(meta.members, None);
        // params stringified via render_value: typed Scalars keep their lexical form.
        assert_eq!(
            meta.params,
            vec![("len".to_string(), "10".to_string()), ("scale".to_string(), "0.5".to_string())]
        );
    }

    /// #99: a sweep/walk-forward family-member stdout line embeds the `RunReport` in
    /// its own declaration key order (manifest leads with `commit`), byte-matching the
    /// stored `families.jsonl` — never `serde_json::Value`'s alphabetical order (which
    /// would lead the manifest with `broker`).
    #[test]
    fn family_member_line_keeps_report_in_store_key_order() {
        let report = RunReport {
            manifest: sim_optimal_manifest(vec![], (Timestamp(0), Timestamp(0)), 0, 1.0),
            metrics: summarize(&[], &[]),
        };
        let line = family_member_line("demo-1", &report);
        assert!(
            line.starts_with(r#"{"family_id":"demo-1","report":{"manifest":{"commit":"#),
            "got: {line}"
        );
        assert!(
            !line.contains(r#""manifest":{"broker":"#),
            "manifest re-alphabetized (broker-first), should be commit-first: {line}"
        );
    }

    /// #99: the Monte-Carlo per-draw line carries the `seed` between `family_id` and
    /// `report`, and the embedded report stays in store (commit-first) key order.
    #[test]
    fn mc_member_line_keeps_report_in_store_key_order_with_seed() {
        let report = RunReport {
            manifest: sim_optimal_manifest(vec![], (Timestamp(0), Timestamp(0)), 7, 1.0),
            metrics: summarize(&[], &[]),
        };
        let line = mc_member_line("mc-1", 7, &report);
        assert!(
            line.starts_with(r#"{"family_id":"mc-1","seed":7,"report":{"manifest":{"commit":"#),
            "got: {line}"
        );
        assert!(
            !line.contains(r#""manifest":{"broker":"#),
            "manifest re-alphabetized (broker-first), should be commit-first: {line}"
        );
    }

    #[test]
    fn data_source_synthetic_pip_and_window_match_the_built_ins() {
        let env = project::Env::std();
        let d = DataSource::Synthetic;
        assert_eq!(d.pip_size(), SYNTHETIC_PIP_SIZE);
        assert!(!d.run_sources(&env, &[aura_ingest::M1Field::Close]).is_empty());
        assert_eq!(d.wf_window_sizes(), (24, 12, 12));
        // full_window equals window_of over the showcase stream (byte-unchanged source)
        let s: Vec<Box<dyn aura_engine::Source>> = vec![Box::new(VecSource::new(showcase_prices()))];
        assert_eq!(d.full_window(&env), window_of(&s).unwrap());
    }

    /// A multi-column blueprint over synthetic data (a single close walk)
    /// refuses honestly through the builders' exit-free Err contract, naming
    /// the beyond-close columns and the --real remedy — never a panic from a
    /// source-count mismatch. High/low-consuming, closed (mc requires it),
    /// with the mandatory `bias` output.
    const OHLC_REFUSAL_BLUEPRINT: &str = r#"{
  "format_version": 1,
  "blueprint": {
    "name": "hl_range",
    "nodes": [ {"primitive":{"type":"Sub"}} ],
    "edges": [],
    "input_roles": [
      {"name":"high","targets":[{"node":0,"slot":0}],"source":"F64"},
      {"name":"low","targets":[{"node":0,"slot":1}],"source":"F64"}
    ],
    "output": [{"node":0,"field":0,"name":"bias"}]
  }
}"#;

    #[test]
    fn synthetic_data_refuses_a_multi_column_blueprint() {
        let env = project::Env::std();
        let err = blueprint_mc_family(OHLC_REFUSAL_BLUEPRINT, 2, &DataSource::Synthetic, &env)
            .expect_err("a high/low blueprint cannot run over the synthetic close walk");
        assert_eq!(
            err,
            "strategy \"hl_range\" consumes columns beyond close (high, low) — synthetic \
             data generates a close series only; run with --real <SYMBOL>"
        );
        let err = blueprint_sweep_family(
            OHLC_REFUSAL_BLUEPRINT,
            &[("x".to_string(), vec![Scalar::i64(1)])],
            &DataSource::Synthetic,
            &env,
        )
        .expect_err("the synthetic sweep path refuses the same shape");
        assert!(err.contains("consumes columns beyond close"), "got: {err}");
    }

    #[test]
    fn wf_real_roller_sizes_are_90_30_30_days_in_ns() {
        // Independent expected value: a day reconstructed from its time units
        // (24 h * 60 min * 60 s * 1e9 ns), not the constant's own `86_400_000_000_000`
        // literal — so the test fails if either the literal or the day-count is wrong.
        let day_ns: i64 = 24 * 60 * 60 * 1_000_000_000;
        assert_eq!(WF_REAL_IS_NS, 90 * day_ns);
        assert_eq!(WF_REAL_OOS_NS, 30 * day_ns);
        assert_eq!(WF_REAL_STEP_NS, 30 * day_ns);
    }

    /// Property: a window that already fits the fixed 90/30/30-day roller passes
    /// it through byte-identical (the year-plus anchor/e2e grade pins rely on this
    /// branch never perturbing the fixed sizes).
    #[test]
    fn fit_wf_ms_sizes_passes_through_when_the_window_already_fits() {
        let day_ms: i64 = 24 * 60 * 60 * 1_000;
        let from_ms = 0;
        let to_ms = 121 * day_ms; // > 90 + 30 days
        assert_eq!(fit_wf_ms_sizes(from_ms, to_ms), wf_ms_sizes());
    }

    /// Property: a window shorter than IS+OOS scales the roller DOWN to the
    /// window, preserving the fixed 3:1 IS:OOS ratio, with `step_ms == oos_ms`
    /// (one roll over the fit window) and `is_ms + oos_ms <= span_ms` always.
    #[test]
    fn fit_wf_ms_sizes_scales_down_to_a_short_window_at_3_to_1() {
        let day_ms: i64 = 24 * 60 * 60 * 1_000;
        let from_ms = 0;
        let to_ms = 30 * day_ms; // far shorter than the fixed 90+30-day roller
        let span_ms = (to_ms - from_ms) as u64;
        let (is_ms, oos_ms, step_ms) = fit_wf_ms_sizes(from_ms, to_ms);
        assert_eq!(is_ms, oos_ms * 3, "the fit preserves the fixed 3:1 IS:OOS ratio");
        assert_eq!(step_ms, oos_ms, "one roll over the fit window: step == oos");
        assert!(is_ms + oos_ms <= span_ms, "the fit roller must fit inside the window");
        assert!(oos_ms > 0, "a 30-day window must yield a non-degenerate fit");
    }

    #[test]
    fn sim_optimal_manifest_renders_per_instrument_pip() {
        let m = sim_optimal_manifest(vec![], (Timestamp(1), Timestamp(2)), 0, 1.0);
        assert_eq!(m.broker, "sim-optimal(pip_size=1)");
        let m2 = sim_optimal_manifest(vec![], (Timestamp(1), Timestamp(2)), 0, 0.0001);
        assert_eq!(m2.broker, "sim-optimal(pip_size=0.0001)");
    }

    // Note (#159 cut 4 collateral): this also removes the `member_key_*` unit
    // tests, the `pair()` fixture helper they shared, and
    // `momentum_param_space_is_ema_exposure_longonly` /
    // `momentum_sweep_is_deterministic_and_has_eight_points` — every one of them
    // exercised `member_key`/`momentum_blueprint_with_sinks`/`momentum_sweep_family`,
    // whose only production caller was the retired PIP built-in `run_sweep`
    // machinery (Task 1 of this iter deleted `member_key`/`MAX_KEY`/`fnv1a64`
    // outright: their sole callers were `sweep_family`/`momentum_sweep_family`,
    // both gone). The member-key-distinctness assertion in the survivor test
    // below is dropped for the same reason (`member_key` no longer exists); the
    // doc comment is trimmed to match.

    /// Property: a `blueprint_sweep_family` member built from a serialized signal is
    /// the SAME trading result as the cycle-1 single run of that signal at the same
    /// params — the loaded-blueprint sweep reuses the identical `wrap_r` run path
    /// (the keystone). Every member of one family carries the SAME `topology_hash` (the
    /// loaded signal's, the deviation from the Rust-built mirror).
    #[test]
    fn blueprint_sweep_member_equals_single_run_and_shares_topology_hash() {
        // An OPEN signal (both SMA knobs free) so the sweep can bind them by name; the
        // serialized doc round-trips to the topology the single run hashes.
        let env = project::Env::std();
        let open = load_open_r_sma();
        let doc = blueprint_to_json(&open).expect("serializes");
        let data = DataSource::Synthetic;
        // fast pinned at 2, slow varied over {4, 6}: a 2x1 grid, slow the varying axis.
        let axes = vec![
            ("sma_signal.fast.length".to_string(), vec![Scalar::i64(2)]),
            ("sma_signal.slow.length".to_string(), vec![Scalar::i64(4), Scalar::i64(6)]),
        ];
        let family = blueprint_sweep_family(&doc, &axes, &data, &env).expect("named axes resolve");
        assert_eq!(family.points.len(), 2, "2x1 grid -> 2 members");

        // (b) every member carries the shared topology_hash of the loaded signal.
        let topo = topology_hash(&open);
        assert_eq!(topo.len(), 64, "topology_hash is a 64-hex SHA256");
        for pt in &family.points {
            assert_eq!(pt.report.manifest.topology_hash.as_deref(), Some(topo.as_str()));
        }

        // (a) the slow=4 member reproduces the cycle-1 single run at fast=2, slow=4 —
        // same equity/exposure stream (total_pips/max_drawdown/bias_sign_flips) and the
        // same topology_hash, proving the loaded blueprint runs through the identical path.
        let single = run_signal_r(
            load_open_r_sma(),
            &[Scalar::i64(2), Scalar::i64(4)],
            RunData::Synthetic,
            0,
            &env,
        );
        let member4 = &family.points[0].report; // slow=4 is the first odometer point
        assert_eq!(member4.metrics, single.metrics, "loaded sweep member == single run");
        assert_eq!(member4.manifest.topology_hash, single.manifest.topology_hash);
    }

    #[test]
    fn blueprint_axis_probe_lists_prefixed_open_knobs() {
        // The open fixture's two SMA lengths are the sweepable knobs; the probe
        // wraps the signal (name "sma_signal") so the names are prefixed —
        // exactly what `--axis` binds.
        let env = project::Env::std();
        let open = include_str!("../tests/fixtures/r_sma_open.json");
        let space = blueprint_axis_probe(open, &env).param_space();
        let names: Vec<&str> = space.iter().map(|p| p.name.as_str()).collect();
        assert_eq!(names, ["sma_signal.fast.length", "sma_signal.slow.length"]);
        assert!(space.iter().all(|p| matches!(p.kind, ScalarKind::I64)));

        // A closed blueprint (both lengths bound) has no open axes.
        let closed = include_str!("../examples/r_sma.json");
        assert!(blueprint_axis_probe(closed, &env).param_space().is_empty());
    }

    #[test]
    fn blueprint_walkforward_family_refits_each_window() {
        // The closed blueprint's two bound SMA lengths are re-fit per IS window over
        // a 2x2 grid via the #246 bound-override reopen path (bound = overridable default).
        let env = project::Env::std();
        let doc = include_str!("../examples/r_sma.json");
        let axes = vec![
            ("sma_signal.fast.length".to_string(), vec![Scalar::i64(2), Scalar::i64(3)]),
            ("sma_signal.slow.length".to_string(), vec![Scalar::i64(4), Scalar::i64(6)]),
        ];
        let result = blueprint_walkforward_family(doc, &axes, &DataSource::Synthetic, Selection::Argmax, &env);
        // 24/12/12 over the 60-bar synthetic span -> 3 rolling windows.
        assert_eq!(result.windows.len(), 3, "three rolling IS/OOS windows");
        for w in &result.windows {
            assert_eq!(w.run.chosen_params.len(), 2, "both axes re-fit each window");
            assert!(w.run.oos_report.metrics.r.is_some(), "OOS record is R-metrics");
        }
        // reduce-mode retains no raw pip curve -> the stitched pip-equity is empty.
        assert!(result.stitched_oos_equity.is_empty(), "no raw pip curve in reduce-mode");
    }

    #[test]
    fn blueprint_mc_family_seeds_differ() {
        // MC over a CLOSED signal (both SMA knobs bound): 3 seeds -> 3 draws, one shared
        // topology_hash, and DIFFERING metrics — the seed reaches the DATA (a distinct
        // synthetic walk per draw), not just the manifest label. The anti-degenerate guard:
        // a regression to seed-as-label-only would make the three draws identical.
        let env = project::Env::std();
        let closed = load_closed_r_sma();
        let doc = blueprint_to_json(&closed).expect("serializes");
        let family = blueprint_mc_family(&doc, 3, &DataSource::Synthetic, &env).expect("closed blueprint");

        assert_eq!(family.draws.len(), 3, "one draw per seed");
        assert_eq!(family.draws.iter().map(|d| d.seed).collect::<Vec<_>>(), vec![1, 2, 3]);
        let topo = family.draws[0].report.manifest.topology_hash.clone();
        assert!(topo.is_some(), "members carry a topology_hash");
        assert!(
            family.draws.iter().all(|d| d.report.manifest.topology_hash == topo),
            "all members share one topology_hash"
        );
        let m: Vec<_> = family.draws.iter().map(|d| &d.report.metrics).collect();
        assert!(m[0] != m[1] || m[1] != m[2], "seeds must yield differing realizations");
    }

    #[test]
    fn blueprint_mc_family_rejects_vacuous_deep_lookback() {
        // Property: the mc family builder REFUSES a silent-vacuous Monte-Carlo — one where
        // every per-seed draw collapses to a bit-identical realization — by RETURNING a named
        // error rather than an `Ok` family that looks like a real (but indistinguishable)
        // distribution. A CLOSED deep-lookback signal whose slow SMA length (60) equals the
        // fixed 60-bar synthetic walk never warms, so every seed yields zero trades and thus
        // identical metrics; that is a wrong result with no error (C10 refuse-don't-guess).
        let env = project::Env::std();
        // slow len (60) == walk len -> never warms
        let mut g = GraphBuilder::new("deep_probe");
        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(60)));
        let spread = g.add(Sub::builder());
        let exposure = g.add(Bias::builder().named("bias").bind("scale", Scalar::f64(0.5)));
        let price = g.source_role("price", ScalarKind::F64);
        g.feed(price, vec![fast.input("series"), slow.input("series")]);
        g.connect(fast.output("value"), spread.input("lhs"));
        g.connect(slow.output("value"), spread.input("rhs"));
        g.connect(spread.output("value"), exposure.input("signal"));
        g.expose(exposure.output("bias"), "bias");
        let deep = g.build().expect("deep probe wiring resolves");
        let doc = blueprint_to_json(&deep).expect("serializes");
        let err = blueprint_mc_family(&doc, 3, &DataSource::Synthetic, &env)
            .expect_err("a vacuous (all-identical) Monte-Carlo is rejected, not returned");
        assert!(
            err.contains("vacuous") || err.contains("identical"),
            "names the vacuous/degenerate condition: {err}"
        );
    }

    #[test]
    fn blueprint_mc_family_rejects_an_open_blueprint() {
        // Property: the mc family builder REFUSES an open blueprint (free knobs) by RETURNING
        // a named error — never a hidden process exit — so the closed-blueprint precondition
        // is unit-testable (the IO wrapper renders it to stderr + exit 2, mirroring the sibling
        // blueprint_sweep_family). MC binds no axis, so a free knob would have no binder; the
        // rejection pre-empts the downstream compile_with_params arity panic.
        let env = project::Env::std();
        // both SMA knobs free -> non-empty param_space
        let open = load_open_r_sma();
        let doc = blueprint_to_json(&open).expect("serializes");
        let err = blueprint_mc_family(&doc, 4, &DataSource::Synthetic, &env)
            .expect_err("an open blueprint is rejected, not run");
        assert!(err.contains("closed blueprint"), "names the closed-blueprint requirement: {err}");
    }

    #[test]
    fn reproduce_family_re_derives_every_member_bit_identically() {
        // a unique temp runs store so the on-disk family + blueprint store do not collide.
        let dir = std::path::Path::new(concat!(env!("CARGO_MANIFEST_DIR"), "/../../target/tmp"))
            .join("aura-repro");
        let _ = std::fs::remove_dir_all(&dir);
        std::fs::create_dir_all(&dir).expect("temp dir");
        let reg = Registry::open(dir.join("runs.jsonl"));

        let env = project::Env::std();
        let open = load_open_r_sma();
        let doc = blueprint_to_json(&open).expect("serializes");
        let data = DataSource::Synthetic;
        // 2x grid over slow.length {4,6} at fast=2 — slow=4 is the open-at-end member.
        let axes = vec![
            ("sma_signal.fast.length".to_string(), vec![Scalar::i64(2)]),
            ("sma_signal.slow.length".to_string(), vec![Scalar::i64(4), Scalar::i64(6)]),
        ];
        let family = blueprint_sweep_family(&doc, &axes, &data, &env).expect("axes resolve");

        // persist exactly as run_blueprint_sweep does: store the blueprint, append the family.
        let topo = family.points[0].report.manifest.topology_hash.clone().expect("topo");
        let canonical =
            blueprint_to_json(&blueprint_from_json(&doc, &|t| std_vocabulary(t)).unwrap()).unwrap();
        reg.put_blueprint(&topo, &canonical).expect("store blueprint");
        let id = reg
            .append_family("repro", FamilyKind::Sweep, &sweep_member_reports(&family))
            .expect("append");

        // reproduce: every member re-derives bit-identically (incl the open-at-end member).
        let rep = reproduce_family_in(&reg, &id, &data, &env);
        assert_eq!(rep.outcomes.len(), 2, "two members reproduced");
        assert!(
            rep.outcomes.iter().all(|(_, ok)| *ok),
            "every member re-derives bit-identically: {:?}",
            rep.outcomes
        );

        let _ = std::fs::remove_dir_all(&dir);
    }

    #[test]
    fn reproduce_re_derives_the_member_stop_regime_not_the_default() {
        // Property: reproduce re-runs each member under the SAME stop regime the member
        // was minted with (its manifest stamps stop_length/stop_k) — the stop defines
        // the risk unit R and is part of the member's identity (C1). A family minted
        // under a NON-default vol-stop regime (a campaign risk-regime cell, or wf/mc/
        // generalize with --stop-length/--stop-k) must still reproduce bit-identically;
        // silently re-running it under the default Vol{3,2.0} reports a spurious DIVERGED.
        let dir = std::path::Path::new(concat!(env!("CARGO_MANIFEST_DIR"), "/../../target/tmp"))
            .join("aura-repro-stop");
        let _ = std::fs::remove_dir_all(&dir);
        std::fs::create_dir_all(&dir).expect("temp dir");
        let reg = Registry::open(dir.join("runs.jsonl"));

        let env = project::Env::std();
        let data = DataSource::Synthetic;
        // A CLOSED blueprint (both SMA knobs bound): its wrapped param_space is empty, so
        // the stop regime is the ONLY dimension that can differ between mint and reproduce
        // — isolating the defect to the hardcoded reproduce-side stop.
        let closed = load_closed_r_sma();
        let doc = blueprint_to_json(&closed).expect("serializes");
        let reload = || blueprint_from_json(&doc, &|t| std_vocabulary(t)).expect("loads");
        let topo = topology_hash(&reload());
        let space = blueprint_axis_probe(&doc, &env).param_space();
        let pip = data.pip_size();
        let window = data.full_window(&env);

        // Mint one member under a NON-default vol-stop regime (default is length=3, k=2.0).
        // run_blueprint_member stamps stop_length=8/stop_k=4.0 into the manifest params.
        let non_default = StopRule::Vol { length: 8, k: 4.0 };
        let binding = binding::resolve_binding("stoprepro", reload().input_roles(), &BTreeMap::new())
            .expect("the price role resolves");
        let report = run_blueprint_member(
            reload(),
            &[],
            &space,
            data.run_sources(&env, &binding.columns()),
            window,
            0,
            pip,
            &topo,
            &env,
            non_default,
            &binding,
            &[],
            "GER40",
        );

        // persist exactly as the sweep/campaign paths do: store the blueprint, append the family.
        let canonical = blueprint_to_json(&reload()).unwrap();
        reg.put_blueprint(&topo, &canonical).expect("store blueprint");
        let id = reg
            .append_family("stoprepro", FamilyKind::Sweep, &[report])
            .expect("append");

        // reproduce: the member re-derives bit-identically only if reproduce honours the
        // manifest's recorded stop regime — currently DIVERGED because reproduce hardcodes
        // the default Vol{3,2.0} for both the param-space probe and the member re-run.
        let rep = reproduce_family_in(&reg, &id, &data, &env);
        assert_eq!(rep.outcomes.len(), 1, "one member reproduced");
        assert!(
            rep.outcomes.iter().all(|(_, ok)| *ok),
            "the non-default-stop member re-derives bit-identically: {:?}",
            rep.outcomes
        );

        let _ = std::fs::remove_dir_all(&dir);
    }

    /// #234: a family minted under a cost model must reproduce bit-identically —
    /// reproduce re-derives the components from the member manifest (the #233
    /// stop-regime pattern). Without the re-derivation the re-run joins an
    /// empty cost slice, its net_expectancy_r reverts to gross, and the member
    /// reports DIVERGED. All three variants ride along so every knob name
    /// round-trips (the vol_slippage component also exercises the reduce-mode
    /// vol proxy).
    #[test]
    fn reproduce_family_re_derives_a_costed_member_bit_identically() {
        let dir = std::path::Path::new(concat!(env!("CARGO_MANIFEST_DIR"), "/../../target/tmp"))
            .join("aura-repro-cost");
        let _ = std::fs::remove_dir_all(&dir);
        std::fs::create_dir_all(&dir).expect("temp dir");
        let reg = Registry::open(dir.join("runs.jsonl"));

        let env = project::Env::std();
        let data = DataSource::Synthetic;
        let doc = blueprint_to_json(&load_closed_r_sma()).expect("serializes");
        let reload = || blueprint_from_json(&doc, &|t| std_vocabulary(t)).expect("loads");
        let topo = topology_hash(&reload());
        let space = blueprint_axis_probe(&doc, &env).param_space();
        let pip = data.pip_size();
        let window = data.full_window(&env);
        let binding = binding::resolve_binding("costrepro", reload().input_roles(), &BTreeMap::new())
            .expect("the price role resolves");
        let cost = vec![
            aura_research::CostSpec::Constant {
                cost_per_trade: aura_research::CostValue::Scalar(0.0005),
            },
            aura_research::CostSpec::VolSlippage {
                slip_vol_mult: aura_research::CostValue::Scalar(0.5),
            },
            aura_research::CostSpec::Carry {
                carry_per_cycle: aura_research::CostValue::Scalar(0.0001),
            },
        ];
        let report = run_blueprint_member(
            reload(),
            &[],
            &space,
            data.run_sources(&env, &binding.columns()),
            window,
            0,
            pip,
            &topo,
            &env,
            StopRule::Vol { length: R_SMA_STOP_LENGTH, k: R_SMA_STOP_K },
            &binding,
            &cost,
            "GER40",
        );
        // Non-vacuity: the cost model must actually bite, else a DIVERGED
        // verdict could never be observed and this pin proves nothing.
        let r = report.metrics.r.as_ref().expect("member carries R metrics");
        assert_ne!(r.net_expectancy_r, r.expectancy_r, "the cost model must move net off gross");

        let canonical = blueprint_to_json(&reload()).unwrap();
        reg.put_blueprint(&topo, &canonical).expect("store blueprint");
        let id = reg.append_family("costrepro", FamilyKind::Sweep, &[report]).expect("append");

        let rep = reproduce_family_in(&reg, &id, &data, &env);
        assert_eq!(rep.outcomes.len(), 1, "one member reproduced");
        assert!(
            rep.outcomes.iter().all(|(_, ok)| *ok),
            "the costed member re-derives bit-identically: {:?}",
            rep.outcomes
        );
        let _ = std::fs::remove_dir_all(&dir);
    }

    #[test]
    fn reproduce_family_re_derives_every_mc_member_bit_identically() {
        let dir = std::path::Path::new(concat!(env!("CARGO_MANIFEST_DIR"), "/../../target/tmp"))
            .join("aura-repro-mc");
        let _ = std::fs::remove_dir_all(&dir);
        std::fs::create_dir_all(&dir).expect("temp dir");
        let reg = Registry::open(dir.join("runs.jsonl"));

        // a CLOSED signal (both SMA knobs bound) — MC binds no axis.
        let env = project::Env::std();
        let closed = load_closed_r_sma();
        let doc = blueprint_to_json(&closed).expect("serializes");
        let data = DataSource::Synthetic;
        let family = blueprint_mc_family(&doc, 3, &data, &env).expect("closed blueprint");

        // persist exactly as run_blueprint_mc does: store the blueprint, append the MC family.
        let topo = family.draws[0].report.manifest.topology_hash.clone().expect("topo");
        let canonical =
            blueprint_to_json(&blueprint_from_json(&doc, &|t| std_vocabulary(t)).unwrap()).unwrap();
        reg.put_blueprint(&topo, &canonical).expect("store blueprint");
        let id = reg
            .append_family("mcrepro", FamilyKind::MonteCarlo, &mc_member_reports(&family))
            .expect("append");

        // reproduce: every MC member re-derives bit-identically (its seed-driven walk is
        // reconstructed from manifest.seed — the realization branch).
        let rep = reproduce_family_in(&reg, &id, &data, &env);
        assert_eq!(rep.outcomes.len(), 3, "three MC members reproduced");
        assert!(
            rep.outcomes.iter().all(|(_, ok)| *ok),
            "every MC member re-derives bit-identically: {:?}",
            rep.outcomes
        );
        let _ = std::fs::remove_dir_all(&dir);
    }

    /// #246: a fully-bound (closed) blueprint IS sweepable — an axis naming a
    /// bound param re-opens it (the bound value is the default), so the retired
    /// "fully bound; nothing to sweep" refusal must not resurface. Inverse
    /// guard: an axis naming NEITHER an open nor a bound param gets the one
    /// clear boundary message (not a terse `UnknownKnob` debug leak).
    #[test]
    fn blueprint_sweep_family_overrides_a_bound_param_and_names_unknown_axes() {
        let env = project::Env::std();
        // same fixture the retired test swept: both SMA knobs bound.
        let closed = load_closed_r_sma();
        let doc = blueprint_to_json(&closed).expect("serializes");

        // (a) override axis: two members, no refusal
        let axes = vec![(
            "sma_signal.fast.length".to_string(),
            vec![Scalar::i64(2), Scalar::i64(4)],
        )];
        let fam = blueprint_sweep_family(&doc, &axes, &DataSource::Synthetic, &env)
            .expect("a bound param is a default — the axis overrides it");
        assert_eq!(fam.points.len(), 2);

        // (b) unknown axis: the boundary message, no UnknownKnob leak
        let bad = vec![("sma_signal.nope".to_string(), vec![Scalar::i64(1)])];
        let err = blueprint_sweep_family(&doc, &bad, &DataSource::Synthetic, &env)
            .expect_err("an axis matching neither space is refused");
        assert!(err.contains("names no param"), "boundary message, got: {err}");
        assert!(err.contains("--list-axes"), "must point at --list-axes: {err}");
        assert!(!err.contains("UnknownKnob"), "must not leak the debug render: {err}");
    }

    /// #249: an axis-reopened bound param flows through `params` ("what
    /// varied") and must NOT also appear in `defaults` ("what was held") — the
    /// two are disjoint by construction. Sweeping `fast.length` on the fully
    /// bound r_sma example leaves `slow.length`/`bias.scale` untouched (they
    /// stay in `defaults`), while `fast.length` moves to `params` and drops
    /// out of `defaults` entirely.
    #[test]
    fn sweep_override_excludes_the_reopened_default_from_the_manifest() {
        let env = project::Env::std();
        let closed = load_closed_r_sma();
        let doc = blueprint_to_json(&closed).expect("serializes");
        let axes = vec![("sma_signal.fast.length".to_string(), vec![Scalar::i64(2)])];
        let fam = blueprint_sweep_family(&doc, &axes, &DataSource::Synthetic, &env)
            .expect("a bound param is a default — the axis overrides it");
        assert_eq!(fam.points.len(), 1);
        let manifest = &fam.points[0].report.manifest;

        let param_names: Vec<&str> = manifest.params.iter().map(|(n, _)| n.as_str()).collect();
        assert!(param_names.contains(&"sma_signal.fast.length"), "swept axis is a param: {param_names:?}");

        let default_names: Vec<&str> = manifest.defaults.iter().map(|(n, _)| n.as_str()).collect();
        assert!(
            !default_names.contains(&"sma_signal.fast.length"),
            "the reopened default must not also appear in defaults: {default_names:?}"
        );
        assert_eq!(
            default_names,
            ["sma_signal.slow.length", "sma_signal.bias.scale"],
            "the untouched bound params stay in defaults: {default_names:?}"
        );
    }

    /// Property: an MC family's `aura reproduce` lines carry the member's own `seed=<N>`
    /// label. MC members hold no tuning params (the params-join is empty), so the line would
    /// otherwise print a BLANK member label; the seed is each draw's realization identity and
    /// must show. Sweep / walk-forward labels still echo their params (covered elsewhere).
    #[test]
    fn reproduce_mc_member_labels_carry_the_seed() {
        let dir = std::path::Path::new(concat!(env!("CARGO_MANIFEST_DIR"), "/../../target/tmp"))
            .join("aura-repro-mc-seed");
        let _ = std::fs::remove_dir_all(&dir);
        std::fs::create_dir_all(&dir).expect("temp dir");
        let reg = Registry::open(dir.join("runs.jsonl"));

        let env = project::Env::std();
        // MC binds no axis -> closed blueprint
        let closed = load_closed_r_sma();
        let doc = blueprint_to_json(&closed).expect("serializes");
        let data = DataSource::Synthetic;
        let family = blueprint_mc_family(&doc, 3, &data, &env).expect("closed blueprint");

        let topo = family.draws[0].report.manifest.topology_hash.clone().expect("topo");
        let canonical =
            blueprint_to_json(&blueprint_from_json(&doc, &|t| std_vocabulary(t)).unwrap()).unwrap();
        reg.put_blueprint(&topo, &canonical).expect("store blueprint");
        let id = reg
            .append_family("mcseed", FamilyKind::MonteCarlo, &mc_member_reports(&family))
            .expect("append");

        let rep = reproduce_family_in(&reg, &id, &data, &env);
        assert_eq!(rep.outcomes.len(), 3, "three MC members");
        for (label, _) in &rep.outcomes {
            assert!(
                label.starts_with("seed="),
                "an MC reproduce label carries the seed, not a blank params-join: {label:?}"
            );
        }
        let seen: HashSet<&str> = rep.outcomes.iter().map(|(l, _)| l.as_str()).collect();
        assert!(
            seen.contains("seed=1") && seen.contains("seed=2") && seen.contains("seed=3"),
            "each MC draw's own seed appears: {seen:?}"
        );
        let _ = std::fs::remove_dir_all(&dir);
    }

    /// Property: an f64 blueprint param survives the content-addressed store's
    /// serialize -> parse -> re-serialize round-trip **bit-identically**, so `aura
    /// reproduce` re-derives an f64-bearing member without DIVERGED. This is exactly
    /// what workspace `serde_json/float_roundtrip` buys: the constant below is a
    /// full-precision f64 (`0.12387080150408619`) that the DEFAULT serde_json parser
    /// mis-parses by 1 ULP — with `float_roundtrip` on it parses back exactly, so the
    /// canonical bytes are stable and the re-run reproduces. The i64-axis reproduce
    /// tests exercise a `scale=0.5` blueprint (exactly representable), so this is the
    /// only test that actually depends on the feature.
    #[test]
    fn f64_blueprint_param_survives_store_round_trip_bit_identically() {
        // 1-ULP canary for serde_json float_roundtrip: parses back off-by-one without it.
        const HARD_SCALE: f64 = 0.12387080150408619;
        // A one-node signal whose only knob is the non-short-decimal f64 scale.
        let mut g = GraphBuilder::new("scale_probe");
        let bias = g.add(Bias::builder().named("bias").bind("scale", Scalar::f64(HARD_SCALE)));
        let signal = g.source_role("signal", ScalarKind::F64);
        g.feed(signal, vec![bias.input("signal")]);
        g.expose(bias.output("bias"), "bias");
        let sig = g.build().expect("one-node bias signal wiring resolves");

        // the store keeps canonical bytes verbatim (dumb bytes-by-key), so the f64
        // fidelity lives entirely in this serialize -> parse -> re-serialize hop.
        let doc = blueprint_to_json(&sig).expect("serializes");
        assert!(
            doc.contains("0.12387080150408619"),
            "canonical JSON carries the full-precision f64: {doc}"
        );
        let reloaded = blueprint_from_json(&doc, &|t| std_vocabulary(t)).expect("loads");
        let doc2 = blueprint_to_json(&reloaded).expect("re-serializes");
        // Without float_roundtrip the reparsed scale drifts 1 ULP and ryu re-serializes
        // it to a different string, so these canonical bytes would differ.
        assert_eq!(doc, doc2, "f64 blueprint param survives the store round-trip bit-identically");
    }

    #[test]
    fn parse_param_cells_decodes_typed_cells_in_order_and_refuses_malformed() {
        // The property: `--params` round-trips the externally-tagged Scalar wire form in
        // array order, and a malformed array is refused with the flag named — never
        // silently dropped (a dropped cell would compile-bind a different graph).
        let cells = parse_param_cells("[{\"I64\":2},{\"F64\":0.5}]").expect("valid cell array");
        assert_eq!(cells, vec![Scalar::I64(2), Scalar::F64(0.5)]);
        assert!(parse_param_cells("[]").expect("empty array").is_empty());
        let err = parse_param_cells("{not an array}").unwrap_err();
        assert!(err.contains("--params"), "a malformed --params value names the flag: {err}");
    }

    /// `topology_hash` is deterministic per signal and distinguishes topologies —
    /// the #158 reproducibility-anchor property.
    #[test]
    fn topology_hash_is_stable_and_distinguishes() {
        let closed = load_closed_r_sma();
        let open = load_open_r_sma();
        let h = topology_hash(&closed);
        assert_eq!(h, topology_hash(&closed), "same signal -> same hash");
        assert_ne!(h, topology_hash(&open), "distinct blueprints -> distinct hash");

        // Same shape as the closed example, differing only in the BOUND fast-SMA
        // length (3 vs 2): the hash must still discriminate on bound param values,
        // not merely on topology shape — distinct sweep/mc members must key to
        // distinct store entries.
        let mut g = GraphBuilder::new("sma_signal");
        let fast = g.add(Sma::builder().named("fast").bind("length", Scalar::i64(3)));
        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("bias").bind("scale", Scalar::f64(0.5)));
        let price = g.source_role("price", ScalarKind::F64);
        g.feed(price, vec![fast.input("series"), slow.input("series")]);
        g.connect(fast.output("value"), spread.input("lhs"));
        g.connect(slow.output("value"), spread.input("rhs"));
        g.connect(spread.output("value"), exposure.input("signal"));
        g.expose(exposure.output("bias"), "bias");
        let bound_variant = g.build().expect("bound-value probe wiring resolves");
        assert_ne!(
            h,
            topology_hash(&bound_variant),
            "same shape, different bound value -> different hash"
        );
    }

    /// #158 acc 1 (content-id stability across the store round-trip): a blueprint's content
    /// id survives serialize -> reload -> re-serialize (#164 idempotence) — the property
    /// content-addressed reproduction rests on (the stored bytes re-hash to the members'
    /// topology_hash). `content_id` is the shared primitive `topology_hash` uses.
    #[test]
    fn content_id_is_stable_across_the_store_round_trip() {
        let closed = load_closed_r_sma();
        let json = blueprint_to_json(&closed).expect("serializes");
        let id = content_id(&json);
        assert_eq!(id.len(), 64, "a 64-hex sha256");
        let reloaded = blueprint_from_json(&json, &|t| std_vocabulary(t)).expect("reloads");
        assert_eq!(
            content_id(&blueprint_to_json(&reloaded).expect("re-serializes")),
            id,
            "content id survives serialize -> reload -> re-serialize (#164)"
        );
    }

    /// #158 acc 3 (Tier-1 format addition leaves the content id unchanged): a Tier-1
    /// optional field the blueprint does not use is tolerated by the loader (#156) and
    /// absent from the canonical omit-defaults form, so re-serializing yields the same
    /// bytes and thus the same content id.
    #[test]
    fn content_id_is_stable_across_a_tolerated_tier1_field() {
        let closed = load_closed_r_sma();
        let base = blueprint_to_json(&closed).expect("serializes");
        // a future Tier-1 optional field injected at the top level (the blueprint does not use it).
        let with_extra =
            base.replacen("{\"format_version\":1,", "{\"format_version\":1,\"future_optional\":123,", 1);
        assert_ne!(base, with_extra, "the doc actually carries the extra field");
        let reparsed = blueprint_from_json(&with_extra, &|t| std_vocabulary(t))
            .expect("loader tolerates an unknown Tier-1 field (#156)");
        assert_eq!(
            content_id(&blueprint_to_json(&reparsed).expect("re-serializes")),
            content_id(&base),
            "a Tier-1 optional the blueprint does not use leaves the content id unchanged"
        );
    }

    /// #158 acc 1 (cross-surface agreement): `topology_hash` (from a live `Composite`, the
    /// run/sweep path) and the op-script `graph introspect --content-id` surface are the
    /// SAME hash of the SAME canonical bytes — both go through the one `content_id`
    /// primitive. Pins that the two command paths cannot silently drift apart.
    #[test]
    fn topology_hash_is_the_content_id_of_the_canonical_form() {
        let sig = load_closed_r_sma();
        assert_eq!(topology_hash(&sig), content_id(&blueprint_to_json(&sig).expect("serializes")));
    }

    /// The op-script twin of the closed r-sma example (`load_closed_r_sma`): same
    /// topology — SMA(2)/SMA(4) over one `price` role, spread, Bias with `scale`
    /// BOUND to 0.5 (boundness is identity-bearing) — differing only in debug names
    /// (composite "graph" vs "sma_signal", unnamed Bias vs `.named("bias")`).
    const IDENTITY_TWIN_DOC: &str = r#"[
        {"op":"source","role":"price","kind":"F64"},
        {"op":"add","type":"SMA","name":"fast","bind":{"length":{"I64":2}}},
        {"op":"add","type":"SMA","name":"slow","bind":{"length":{"I64":4}}},
        {"op":"add","type":"Sub"},
        {"op":"add","type":"Bias","bind":{"scale":{"F64":0.5}}},
        {"op":"feed","role":"price","into":["fast.series","slow.series"]},
        {"op":"connect","from":"fast.value","to":"sub.lhs"},
        {"op":"connect","from":"slow.value","to":"sub.rhs"},
        {"op":"connect","from":"sub.value","to":"bias.signal"},
        {"op":"expose","from":"bias.bias","as":"bias"}
    ]"#;

    /// #171 acc 1 (cross-path identity): the Rust `sma_signal` builder and its
    /// op-script twin differ in canonical bytes (debug names) — distinct content
    /// ids — but project to the same identity JSON, hence one identity id across
    /// authoring paths.
    #[test]
    fn identity_id_bridges_the_rust_builder_and_op_script_paths() {
        let rust_built = load_closed_r_sma();
        let env = project::Env::std();
        let json = crate::graph_construct::build_from_str(IDENTITY_TWIN_DOC, &env)
            .expect("op-script twin builds");
        assert_ne!(
            content_id(&json),
            topology_hash(&rust_built),
            "authoring paths keep distinct content ids"
        );
        let loaded = blueprint_from_json(&json, &|t| std_vocabulary(t)).expect("twin reloads");
        let identity = |c: &Composite| {
            content_id(&aura_engine::blueprint_identity_json(c).expect("identity-serializes"))
        };
        assert_eq!(
            identity(&loaded),
            identity(&rust_built),
            "one topology -> one identity id across authoring paths"
        );
    }

    #[test]
    fn mc_r_bootstrap_json_carries_every_bootstrap_field_under_the_mc_r_bootstrap_key() {
        // Property: the `mc_r_bootstrap` output line is the full RBootstrap shape —
        // each field is named and value-faithful, so a renamed/dropped field here
        // breaks a test instead of silently shipping. Pins the user-visible wire
        // shape of the mc R-bootstrap render (the parser + engine primitive are
        // covered elsewhere; this is the output-shape layer).
        let boot = aura_engine::r_bootstrap(&[1.0, -0.5, 2.0, -1.0], 64, 2, 7);
        let line = mc_r_bootstrap_json(&boot);
        let v: serde_json::Value = serde_json::from_str(&line).expect("canonical json line");
        let obj = &v["mc_r_bootstrap"];
        assert_eq!(obj["n_trades"], serde_json::json!(boot.n_trades));
        assert_eq!(obj["block_len"], serde_json::json!(boot.block_len));
        assert_eq!(obj["n_resamples"], serde_json::json!(boot.n_resamples));
        assert_eq!(obj["prob_le_zero"], serde_json::json!(boot.prob_le_zero));
        // e_r is the nested MetricStats block (mean + quantiles), not a flat scalar.
        assert_eq!(obj["e_r"], serde_json::to_value(&boot.e_r).expect("MetricStats serializes"));
        assert!(obj["e_r"]["mean"].is_number(), "e_r should nest the MetricStats block: {line}");
    }

    /// #262 round-trip: a manifest carrying stop_period_minutes re-derives
    /// the VolTf stop; one without it keeps the Vol path — a stored VolTf
    /// member must never silently reproduce under a default Vol stop.
    #[test]
    fn stop_rule_from_params_round_trips_the_vol_tf_stamp() {
        let vol_tf = vec![
            ("stop_period_minutes".to_string(), Scalar::i64(60)),
            ("stop_length".to_string(), Scalar::i64(14)),
            ("stop_k".to_string(), Scalar::f64(2.0)),
        ];
        assert_eq!(
            stop_rule_from_params(&vol_tf),
            StopRule::VolTf { period_minutes: 60, length: 14, k: 2.0 }
        );
        let vol = vec![
            ("stop_length".to_string(), Scalar::i64(3)),
            ("stop_k".to_string(), Scalar::f64(2.0)),
        ];
        assert_eq!(stop_rule_from_params(&vol), StopRule::Vol { length: 3, k: 2.0 });
    }

    /// #262 write-side: `run_blueprint_member` given `StopRule::VolTf` actually
    /// bootstraps and runs the member (the `risk_executor`/`VolTfStop` arm,
    /// end-to-end over a real synthetic realization, not a hand-built value)
    /// and stamps all three knobs into the manifest under the keys
    /// `stop_rule_from_params`'s round-trip above reads back — the write half
    /// of the manifest round-trip pair.
    #[test]
    fn run_blueprint_member_stamps_the_vol_tf_stop_knobs() {
        let env = project::Env::std();
        let data = DataSource::Synthetic;
        let doc = blueprint_to_json(&load_closed_r_sma()).expect("serializes");
        let reload = || blueprint_from_json(&doc, &|t| std_vocabulary(t)).expect("loads");
        let space = blueprint_axis_probe(&doc, &env).param_space();
        let binding = binding::resolve_binding("voltfstamp", reload().input_roles(), &BTreeMap::new())
            .expect("the price role resolves");
        let stop = StopRule::VolTf { period_minutes: 60, length: 3, k: 2.0 };
        let window = data.full_window(&env);
        let pip = data.pip_size();
        let run = crate::run_blueprint_member(
            reload(),
            &[],
            &space,
            data.run_sources(&env, &binding.columns()),
            window,
            0,
            pip,
            "topo",
            &env,
            stop,
            &binding,
            &[],
            "GER40",
        );
        assert!(
            run.manifest.params.contains(&("stop_period_minutes".to_string(), Scalar::i64(60))),
            "manifest must stamp stop_period_minutes: {:?}",
            run.manifest.params
        );
        assert!(
            run.manifest.params.contains(&("stop_length".to_string(), Scalar::i64(3))),
            "manifest must stamp stop_length: {:?}",
            run.manifest.params
        );
        assert!(
            run.manifest.params.contains(&("stop_k".to_string(), Scalar::f64(2.0))),
            "manifest must stamp stop_k: {:?}",
            run.manifest.params
        );
    }

    /// A bare `McCmd` with every optional field defaulted to `None`/empty, so each
    /// `mc_args_from` refusal test below only sets the fields its scenario needs.
    fn bare_mc_cmd() -> McCmd {
        McCmd {
            blueprint: None,
            real: None,
            from: None,
            to: None,
            name: None,
            trace: None,
            axis: Vec::new(),
            stop_length: None,
            stop_k: None,
            block_len: None,
            resamples: None,
            seed: None,
            seeds: None,
        }
    }

    #[test]
    fn mc_args_from_refuses_without_a_real_symbol() {
        let a = bare_mc_cmd();
        let err = mc_args_from(&a).unwrap_err();
        assert!(err.contains("dissolves only over --real"), "refuse message: {err}");
    }

    #[test]
    fn mc_args_from_refuses_an_empty_real_symbol() {
        let a = McCmd { real: Some(String::new()), ..bare_mc_cmd() };
        let err = mc_args_from(&a).unwrap_err();
        assert!(err.contains("dissolves only over --real"), "refuse message: {err}");
    }

    #[test]
    fn mc_args_from_refuses_name_and_trace() {
        let a = McCmd {
            real: Some("GER40".to_string()),
            name: Some("a".to_string()),
            trace: Some("b".to_string()),
            ..bare_mc_cmd()
        };
        let err = mc_args_from(&a).unwrap_err();
        assert!(err.contains("--name/--trace are not accepted"), "refuse message: {err}");
    }

    /// Property (#217): a missing `--stop-length`/`--stop-k` no longer refuses —
    /// each independently defaults to the single-sourced [`R_SMA_STOP_LENGTH`]/
    /// [`R_SMA_STOP_K`] regime (length 3, k 2.0).
    #[test]
    fn mc_args_from_defaults_missing_knobs_to_the_regime() {
        let a = McCmd { real: Some("GER40".to_string()), ..bare_mc_cmd() };
        let (_, _, stop_length, stop_k, ..) = mc_args_from(&a).expect("stop-less mc resolves");
        assert_eq!(stop_length, R_SMA_STOP_LENGTH, "omitted --stop-length defaults to the regime");
        assert_eq!(stop_k, R_SMA_STOP_K, "omitted --stop-k defaults to the regime");
    }

    #[test]
    fn mc_args_from_refuses_a_multi_value_stop() {
        let a = McCmd {
            real: Some("GER40".to_string()),
            stop_length: Some("14,20".to_string()),
            stop_k: Some("2.0".to_string()),
            ..bare_mc_cmd()
        };
        let err = mc_args_from(&a).unwrap_err();
        assert!(err.contains("single risk regime"), "stop-regime refusal: {err}");
    }

    /// Property (#217): a missing `--stop-length`/`--stop-k` no longer refuses —
    /// each independently defaults to the single-sourced [`R_SMA_STOP_LENGTH`]/
    /// [`R_SMA_STOP_K`] regime (length 3, k 2.0).
    #[test]
    fn generalize_args_from_defaults_missing_knobs_to_the_regime() {
        let a = GeneralizeCmd {
            blueprint: Some("candidate.json".to_string()),
            real: Some("GER40,USDJPY".to_string()),
            axis: vec!["k=1".to_string()],
            stop_length: None,
            stop_k: None,
            from: None,
            to: None,
            metric: None,
            name: None,
        };
        let (_, _, stop_length, stop_k, ..) =
            generalize_args_from(&a).expect("stop-less generalize resolves");
        assert_eq!(stop_length, R_SMA_STOP_LENGTH, "omitted --stop-length defaults to the regime");
        assert_eq!(stop_k, R_SMA_STOP_K, "omitted --stop-k defaults to the regime");
    }
}
