Aura/crates/aura-runner/src/member.rs

//! The canonical member-run recipe: harness assembly.
//!
//! The R-scaffolding wrap (`wrap_r`), the single-run and per-member execution
//! paths built on it (`run_signal_r`, `run_blueprint_member`), the loaded-
//! blueprint axis probe (`blueprint_axis_probe`/`blueprint_axis_probe_reopened`)
//! and its #246 override-set machinery (`wrapped_bound_names`,
//! `wrapped_bound_overrides_of`, `override_paths`, `reopen_all`) — the shipped
//! recipe `aura_campaign::MemberRunner` implementors (the CLI's
//! `CliMemberRunner`, and any downstream World program) drive real data
//! through.

use std::collections::{BTreeMap, BTreeSet, HashSet};
use std::sync::{mpsc, Arc, LazyLock};

use aura_composites::{cost_graph, risk_executor, StopRule};
use aura_core::{zip_params, Cell, Firing, ParamSpec, PrimitiveBuilder, Scalar, ScalarKind, Timestamp};
use aura_engine::{
    blueprint_from_json, window_of, BlueprintNode, ColumnarTrace, Composite, GraphBuilder, Harness,
    RunManifest, VecSource,
};
use aura_backtest::{
    summarize, summarize_r, RunMetrics, RunReport, SimBroker, PM_FIELD_NAMES, PM_RECORD_KINDS,
};
use aura_std::{GatedRecorder, LinComb, Recorder, RollingMax, RollingMin, SeriesReducer, Sub};
use aura_strategy::{cost_port, GEOMETRY_WIDTH};

use crate::binding::ResolvedBinding;
use crate::project::Env;
use crate::translate::{R_SMA_STOP_LENGTH, R_SMA_STOP_K};

/// The single build-time commit provenance (`option_env!("AURA_COMMIT")`,
/// falling back to `"unknown"`) — `sim_optimal_manifest`'s `RunManifest.commit`
/// reads this one const, mirroring the CLI shell's own copy (both resolve the
/// same process-wide env var at compile time, so the two cannot disagree).
const ENGINE_COMMIT: &str = match option_env!("AURA_COMMIT") {
    Some(c) => c,
    None => "unknown",
};

/// 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.
pub const SYNTHETIC_PIP_SIZE: f64 = 0.0001;

/// 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)]
pub enum RunData {
    Synthetic,
    Real { symbol: String, from: Option<i64>, to: Option<i64> },
}

/// 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.
pub 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.
pub 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()
}

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

/// 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).
pub 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()
}

/// No-local-data refusal — stderr + exit(1).
pub fn no_real_data(symbol: &str, env: &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: &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 the CLI shell's
/// `DataSource::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.
pub fn pip_or_refuse(server: &Arc<data_server::DataServer>, symbol: &str, env: &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 the CLI shell's `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.
pub fn probe_window(
    server: &Arc<data_server::DataServer>,
    symbol: &str,
    from_ms: Option<i64>,
    to_ms: Option<i64>,
    env: &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: &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 = 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)
}

/// 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 `translate::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).
pub struct CostLeg {
    pub nodes: Vec<PrimitiveBuilder>,
    pub tx_cost: mpsc::Sender<(Timestamp, Vec<Scalar>)>,
    pub tx_net: mpsc::Sender<(Timestamp, Vec<Scalar>)>,
}

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

/// 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)]
pub 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: &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, crate::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 = crate::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).
pub fn key_supply(
    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)]
pub fn resolve_run_data(
    data: &RunData,
    env: &Env,
    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)]
pub fn run_signal_r(
    signal: Composite, params: &[Scalar], data: RunData, seed: u64, env: &Env,
) -> RunReport {
    // topology_hash's own two-line body, inlined: `content_id_of` over the
    // canonical (#164) blueprint JSON — the CLI shell's `topology_hash`
    // helper is the same primitive, kept single-sourced at `aura_research`.
    let topo = aura_research::content_id_of(
        &aura_engine::blueprint_to_json(&signal).expect("a buildable signal serializes"),
    ); // 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 = crate::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: {}", crate::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(&aura_engine::f64_field(&eq_rows, 0), &aura_engine::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 }
}

/// 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)]
pub 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: &Env,
    stop: StopRule,
    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: crate::translate::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
    // `translate::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 `translate::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) = crate::translate::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`.
pub fn blueprint_axis_probe(doc: &str, env: &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).
pub fn blueprint_axis_probe_reopened(doc: &str, env: &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 = crate::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 the
/// CLI shell's `validate_and_register_axes` each independently rebuilt this
/// exact set (the rule-of-three the neighbouring doc comment itself cites).
pub 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
/// `axes::raw_bound_overrides_of`, whose `names` are already RAW (a campaign
/// document's own namespace, #203) and needs no such stripping.
pub 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.
pub 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)
}

/// 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.
pub 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")
    })
}

/// 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 caller — its only role is regenerating + pinning the shipped
/// examples via this module's own tests; production code loads the shipped
/// JSON, not this builder. `#[cfg(test)]`.
#[cfg(test)]
fn r_breakout_signal(channel: Option<i64>) -> Composite {
    use aura_std::{Delay, Gt, Latch};
    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 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)]`: this module's own `wrap_r`/`run_blueprint_member`
/// unit tests below use it as a realistic non-flat-bias fixture, and its only
/// other role is regenerating + pinning the shipped examples; production loads
/// the shipped JSON, not this builder.
#[cfg(test)]
fn r_meanrev_signal(window: Option<i64>, band_k: Option<f64>) -> Composite {
    use aura_std::{Add, Ema, Gt, Latch, Mul, Scale, Sqrt};
    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 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)]`
/// (see `r_breakout_signal`'s doc): its only role is regenerating + pinning
/// the shipped examples via this module's own tests; production loads the
/// shipped JSON, not this builder.
#[cfg(test)]
fn r_channel_signal(channel: Option<i64>) -> Composite {
    use aura_std::{Delay, Gt, Latch};
    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")
}

#[cfg(test)]
mod tests {
    use super::*;
    use aura_engine::blueprint_to_json;
    use aura_strategy::{ConstantCost, VolSlippageCost};
    use aura_vocabulary::std_vocabulary;

    #[test]
    /// 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. `k = 0` collapses the band to the lagging EWMA mean,
    /// isolating direction + latch from the sigma threshold; 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.
    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 = crate::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:?}");
    }

    #[test]
    /// 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`).
    fn run_blueprint_member_computes_pips_at_the_resolved_pip_not_a_hardwired_default() {
        let env = 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 = crate::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,
        );
    }

    #[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)");
    }

    /// 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` fixture — kept honest
    /// by one constant instead of a hand-synced `Some(3)` literal recurring
    /// across those call sites. `examples/r_breakout.json` is CLI-owned
    /// (`crates/aura-cli/examples/`); these tests read it cross-crate (#295).
    const R_BREAKOUT_CHANNEL: i64 = 3;

    /// 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`.
    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`].
    const R_MEANREV_BAND_K: f64 = 2.0;

    /// 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`]).
    const R_CHANNEL_LENGTH: i64 = 3;

    /// Loads the shipped closed r-sma example (fast=2, slow=4 bound) through
    /// the public `blueprint_from_json` path. `examples/r_sma.json` is
    /// CLI-owned (`crates/aura-cli/examples/`, also read directly by `aura
    /// graph`'s no-argument default); this reads the identical file
    /// cross-crate (#295). aura-cli's own `mod tests` in `main.rs` keeps its
    /// own verbatim copy of this loader for the tests that stayed there.
    fn load_closed_r_sma() -> Composite {
        blueprint_from_json(include_str!("../../aura-cli/examples/r_sma.json"), &|t| std_vocabulary(t))
            .expect("loads")
    }

    /// Regenerates the shipped r_breakout examples from the carved signal. Run by hand:
    /// `cargo test -p aura-runner 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() {
        // CLI-owned target paths (#295: this builder lives in aura-runner, the
        // fixture it regenerates lives in aura-cli) — anchored on
        // `CARGO_MANIFEST_DIR` (this crate's own root), not process cwd, the
        // same pattern `project.rs`'s tmp-dir tests already use.
        let examples_dir = concat!(env!("CARGO_MANIFEST_DIR"), "/../aura-cli/examples");
        std::fs::create_dir_all(examples_dir).expect("examples dir");
        std::fs::write(
            concat!(env!("CARGO_MANIFEST_DIR"), "/../aura-cli/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(
            concat!(env!("CARGO_MANIFEST_DIR"), "/../aura-cli/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!("../../aura-cli/examples/r_breakout.json"),
        );
        assert_eq!(
            blueprint_to_json(&r_breakout_signal(None)).expect("serialize open"),
            include_str!("../../aura-cli/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 = Env::std();
        let loaded = blueprint_from_json(include_str!("../../aura-cli/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 -p aura-runner 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(concat!(env!("CARGO_MANIFEST_DIR"), "/../aura-cli/examples"))
            .expect("examples dir");
        std::fs::write(
            concat!(env!("CARGO_MANIFEST_DIR"), "/../aura-cli/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(
            concat!(env!("CARGO_MANIFEST_DIR"), "/../aura-cli/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 -p aura-runner 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(concat!(env!("CARGO_MANIFEST_DIR"), "/../aura-cli/examples"))
            .expect("examples dir");
        std::fs::write(
            concat!(env!("CARGO_MANIFEST_DIR"), "/../aura-cli/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(
            concat!(env!("CARGO_MANIFEST_DIR"), "/../aura-cli/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!("../../aura-cli/examples/r_channel.json"),
        );
        assert_eq!(
            blueprint_to_json(&r_channel_signal(None)).expect("serialize open"),
            include_str!("../../aura-cli/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 =
                crate::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!("../../aura-cli/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!("../../aura-cli/examples/r_meanrev.json"),
        );
        assert_eq!(
            blueprint_to_json(&r_meanrev_signal(None, None)).expect("open"),
            include_str!("../../aura-cli/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 = Env::std();
        let loaded = blueprint_from_json(include_str!("../../aura-cli/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");
    }

    /// #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 = crate::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 = crate::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 = crate::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 = Env::std();
        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 = crate::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 };
        // The CLI shell's `DataSource::Synthetic` doesn't cross into aura-runner
        // (it's dispatch-layer, #295 out of scope here): this module's own
        // `RunData::Synthetic` + `resolve_run_data` stands in, sourcing
        // `r_sma_prices()` instead of the CLI's `showcase_prices()` — a
        // different but equally non-flat synthetic fixture. Every assertion
        // below is computed from the actual run (never a fixture-specific
        // literal), so the substitution preserves the property under test.
        let sources = || resolve_run_data(&RunData::Synthetic, &env, &binding).0;
        let (_, window, pip) = resolve_run_data(&RunData::Synthetic, &env, &binding);
        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,
                sources(),
                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(sources());
        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"
        );
    }

    /// #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 = Env::std();
        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 = crate::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 };
        // See `run_blueprint_member_joins_a_constant_cost_model_into_net_metrics`'s
        // note: `RunData::Synthetic` + `resolve_run_data` stands in for the
        // CLI-only `DataSource::Synthetic` this test used before the #295 move.
        let (sources, window, pip) = resolve_run_data(&RunData::Synthetic, &env, &binding);
        let run = run_blueprint_member(
            reload(),
            &[],
            &space,
            sources,
            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
        );
    }

}