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, 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, CompileError, 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 aura_campaign::MemberFault;

use crate::binding::ResolvedBinding;
use crate::project::Env;
use crate::tap_plan::{bind_tap_plan, TapPlan};
use crate::translate::{R_SMA_STOP_LENGTH, R_SMA_STOP_K};
use crate::RunnerError;

/// 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: sim_optimal_broker_label(pip_size),
        selection: None,
        instrument: None,
        topology_hash: None,
        project: None,
    }
}

/// The RAW-namespace BOUND-param defaults of `signal` (#249/#328): every
/// `bound_param_space()` entry as a `(<param>, value)` pair, already RAW
/// (`bound_param_space()`'s own path-qualified name, #203 — no wrap-prefix
/// concatenation), 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. Contrast
/// `wrapped_bound_names`, which DOES wrap-prefix (it matches against the
/// WRAPPED open `param_space()`, a different coordinate system than this
/// manifest-recording helper needs).
pub fn raw_bound_defaults(signal: &Composite) -> Vec<(String, Scalar)> {
    signal.bound_param_space().into_iter().map(|b| (b.name, b.value)).collect()
}

/// The plain sim-optimal broker label (no RiskExecutor branch): shared by
/// `sim_optimal_manifest` and (#299) `reproduce_family_in`'s forward-built
/// guard comparison, so the two cannot drift.
pub(crate) fn sim_optimal_broker_label(pip_size: f64) -> String {
    format!("sim-optimal(pip_size={pip_size})")
}

/// 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 (#299: and `reproduce_family_in`'s
/// forward-built guard comparison) so they cannot drift.
pub(crate) 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 (#297: returned, not printed — the CLI/library
/// boundary prints `aura: {message}` and exits with `exit_code`).
pub fn no_real_data(symbol: &str, env: &Env) -> RunnerError {
    RunnerError {
        exit_code: 1,
        message: format!("no local data for symbol '{symbol}' at {}", env.data_path()),
    }
}

/// Empty-in-window refusal (#297: returned, not printed). 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) -> RunnerError {
    let bound = |b: Option<i64>| b.map_or_else(|| "unbounded".to_string(), |v| v.to_string());
    RunnerError {
        exit_code: 1,
        message: format!(
            "no data for symbol '{symbol}' in the requested window [{}, {}] at {}",
            bound(from_ms),
            bound(to_ms),
            env.data_path()
        ),
    }
}

/// Resolve the per-instrument pip from the recorded geometry sidecar, or refuse
/// (#297: returned, not printed) when the symbol has no recorded geometry — the
/// single home of the guessed-pip refusal, shared by `open_real_source` and
/// reproduce's `DataSource::from_choice` chain (`family.rs`). 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) -> Result<f64, RunnerError> {
    match aura_ingest::instrument_geometry(server, symbol) {
        Some(geo) => Ok(geo.pip_size),
        None => Err(RunnerError {
            exit_code: 1,
            message: format!(
                "no recorded geometry for symbol '{symbol}' at {} — \
                 refusing to run a real instrument with a guessed pip",
                env.data_path()
            ),
        }),
    }
}

/// 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,
) -> Result<(Timestamp, Timestamp), RunnerError> {
    aura_ingest::archive_extent(server, std::path::Path::new(&env.data_path()), symbol, from_ms, to_ms)
        .ok_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 returned as a `RunnerError`, #297). Pre-data
/// refusals keep the pip honest by construction. Used by `resolve_run_data`.
#[allow(clippy::type_complexity)]
fn open_real_source(
    symbol: &str,
    from_ms: Option<i64>,
    to_ms: Option<i64>,
    env: &Env,
    fields: &[aura_ingest::M1Field],
) -> Result<(Vec<Box<dyn aura_engine::Source>>, (Timestamp, Timestamp), f64), RunnerError> {
    // 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) {
        return Err(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)
        .ok_or_else(|| no_real_data(symbol, env))?;
    Ok((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::configured(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::configured(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,
) -> Result<
    (
        Vec<Box<dyn aura_engine::Source>>,
        (Timestamp, Timestamp),
        f64,
    ),
    RunnerError,
> {
    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");
            Ok((sources, window, SYNTHETIC_PIP_SIZE))
        }
        RunData::Real { symbol, from, to } => {
            open_real_source(symbol, *from, *to, env, &binding.columns())
        }
    }
}

/// The variant-name string for a `ScalarKind`, matching the wire/CLI spelling
/// used across the codebase (`F64`/`I64`/`Bool`/`Timestamp`).
fn scalar_kind_name(kind: ScalarKind) -> &'static str {
    match kind {
        ScalarKind::F64 => "F64",
        ScalarKind::I64 => "I64",
        ScalarKind::Bool => "Bool",
        ScalarKind::Timestamp => "Timestamp",
    }
}

/// Bug 3 (#319 fieldtest cycle 2): an `--override` value whose lexed kind
/// does not match the param's declared kind (e.g. the bare literal `2` for
/// an F64 param) reaches this compile boundary as a raw
/// `CompileError::ParamKindMismatch { slot, expected, got }` — `slot` is a
/// flat param-space index, meaningless to a caller. `names`/`params` are the
/// SAME order `compile_with_params` was called with (built from `signal`
/// before it was consumed by `wrap_r`, `C11`: composites inline, preserving
/// order), so `slot` indexes both. This names the param path and both kinds,
/// mirroring the campaign leg's own kind-fault prose
/// (`ref_fault_prose`'s `AxisKindMismatch` arm, `research_docs.rs`). Every
/// other `CompileError` variant keeps the existing Debug fallback — out of
/// this bug's scope, and `names`/`params` only reconstruct the one variant
/// that carries an injected value.
fn compile_error_prose(e: &CompileError, names: &[String], params: &[Scalar]) -> String {
    let CompileError::ParamKindMismatch { slot, expected, got } = e else {
        return format!("this blueprint does not compile to a runnable harness: {e:?}");
    };
    let path = names.get(*slot).map(String::as_str).unwrap_or("<unknown>");
    let example = match (expected, params.get(*slot)) {
        (ScalarKind::F64, Some(Scalar::I64(v))) => format!("{v}.0"),
        (ScalarKind::F64, _) => "a decimal literal, e.g. 2.0".to_string(),
        (ScalarKind::I64, _) => "a bare integer literal, e.g. 2".to_string(),
        (ScalarKind::Bool, _) => "true or false".to_string(),
        (ScalarKind::Timestamp, _) => "an integer epoch-ns literal".to_string(),
    };
    format!(
        "--override {path}: expects {}, got {} — write {example}",
        scalar_kind_name(*expected),
        scalar_kind_name(*got)
    )
}

/// 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.
///
/// `topo`: `None` computes `topology_hash` inline from `signal` as always
/// (every existing caller); `Some(hash)` overrides it with the caller's own
/// **reference-semantics** hash (#343, revised) — the exec blueprint leg's
/// `--override` branch passes the loaded base document's content id, computed
/// BEFORE `reopen_all`, so the manifest reaching every consumer inside this
/// function (the record line AND the trace-store persistence below) carries
/// the base document's id, never the reopened topology's own, exactly as the
/// campaign leg's `run_blueprint_member` has always taken its `topo` as a
/// caller-supplied reference (`runner.rs`'s `&cell.strategy_id`).
#[allow(clippy::type_complexity)]
pub fn run_signal_r(
    signal: Composite, params: &[Scalar], data: RunData, seed: u64, env: &Env,
    plan: TapPlan, topo: Option<&str>,
) -> Result<(RunReport, Vec<String>), RunnerError> {
    // #297 fieldtest finding B1 (milestone-safe-to-embed): mirror the CLI's
    // own `exec_blueprint_leg` exposes-neither guard (`aura-cli/src/main.rs`,
    // `has_bias`/`has_tap`) IN the library, before `wrap_r` below wires a
    // "bias" output that does not exist — an embedding host driving
    // `run_signal_r` directly (never through the CLI shell) must see this
    // same typed refusal instead of `wrap_r`'s wiring `.expect` panicking
    // with `UnknownOutPort { name: "bias" }` and killing the host. Same
    // detection (`output()`/`taps()`), same prose family as the CLI's,
    // minus its own "aura: " print prefix — C14 class 2 (argv-named content).
    if !signal.output().iter().any(|o| o.name == "bias") && signal.taps().is_empty() {
        return Err(RunnerError {
            exit_code: 2,
            message: "exec needs either a `bias` output (a strategy) or ≥1 declared tap \
                      (a measurement); this blueprint exposes neither"
                .to_string(),
        });
    }
    // 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`.
    // Skipped when the caller already supplies a reference-semantics hash
    // (`topo: Some(..)`) — no need to hash a signal whose own topology_hash
    // will be overridden anyway.
    let topo: String = match topo {
        Some(t) => t.to_string(),
        None => 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 propagate as a `RunnerError` for the CLI to
    // print and exit on, never a process exit inside the library (#297).
    // C14 class 2: fault in argv-named content (#297)
    let binding = crate::binding::resolve_binding(signal.name(), signal.input_roles(), &BTreeMap::new())
        .map_err(|m| RunnerError { exit_code: 2, message: m })?;
    // C14 class 2: fault in argv-named content (#297)
    if matches!(data, RunData::Synthetic) && !binding.close_only() {
        return Err(RunnerError { exit_code: 2, message: crate::binding::synthetic_refusal(signal.name(), &binding) });
    }
    let names: Vec<String> = signal
        .param_space()
        .iter()
        .map(|p| p.name.clone())
        .collect();
    let defaults = raw_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);
    // C14 class 2: fault in argv-named content (#297)
    let mut flat = wrapped.compile_with_params(params)
        .map_err(|e| RunnerError { exit_code: 2, message: compile_error_prose(&e, &names, params) })?;
    // Bind each declared tap per the plan's subscription, before bootstrap
    // (#283): typed refusals for bad plans, record consumers hold their
    // streaming writer in-graph, folds accumulate O(1), live closures run
    // inline. Dedup stays caller-owned per TapBindError::DuplicateBind's doc.
    // C14 class 2: fault in argv-named content (#297)
    let bound = bind_tap_plan(&mut flat, plan, env, &run_name)
        .map_err(|e| RunnerError { exit_code: e.exit_class(), message: e.to_string() })?;
    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, &[]));
    let skipped = bound.skipped().to_vec();
    // Close the tap plan: drain the ≤1-message channels, write fold rows,
    // then `index.json` — nothing was buffered during the run (#283). A
    // tap-free (or nothing-persisting) plan wrote nothing at all.
    bound.finish(&manifest)
        .map_err(|e| RunnerError { exit_code: e.exit_class(), message: e.to_string() })?;
    Ok((RunReport { manifest, metrics }, skipped))
}

/// 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,
) -> Result<RunReport, MemberFault> {
    let defaults = raw_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 = if cost.is_empty() {
        None
    } else {
        let nodes = crate::translate::cost_nodes_for(cost, instrument)
            .map_err(|f| MemberFault::Bind(f.message()))?;
        Some(CostLeg { nodes, 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
    // One arm's knobs per member (#233 Vol/VolTf, #338 Fixed) — the manifest
    // stamp `stop_rule_from_params` reads back (`translate.rs`), so a
    // campaign's `RiskRegime::Fixed` cell reproduces its own distance, not
    // the baked default vol-stop.
    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(distance) => {
            named.push(("stop_distance".to_string(), Scalar::f64(distance)));
        }
    }
    // 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)
            .map_err(|f| MemberFault::Bind(f.message()))?;
        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));
    Ok(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
/// `aura graph introspect --params`, 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 `aura graph introspect --params`
/// 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 name missing the un-reopened
/// wrapped OPEN space but naming a BOUND param of the strategy — returned in
/// STRATEGY coordinates (the wrap segment 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`/`runner.rs`
/// use over the RECORDED manifest param names (the sweep boundary uses the
/// stricter `override_paths`, which errors on an unmatched name instead).
///
/// #328: matches via [`crate::axes::raw_matches_wrapped`], so `names` may be
/// WRAPPED (every route this cycle leaves untouched) OR RAW (a `Sweep` family
/// minted by `blueprint_sweep_family`, whose manifest now records the raw
/// campaign namespace, #328) — `reproduce_family_in` reproduces families from
/// either route generically, so this helper must tolerate both shapes rather
/// than assume one. Contrast `axes::raw_bound_overrides_of`, whose `names` are
/// ALWAYS RAW (a campaign document's own namespace, #203) by contract.
pub fn wrapped_bound_overrides_of(
    names: &[String],
    open_space: &[ParamSpec],
    signal: &Composite,
) -> Vec<String> {
    let bound = wrapped_bound_names(signal);
    names
        .iter()
        .filter(|n| !open_space.iter().any(|p| crate::axes::raw_matches_wrapped(n, &p.name)))
        .filter_map(|n| {
            bound
                .iter()
                .find(|b| crate::axes::raw_matches_wrapped(n, b))
                .map(|b| crate::axes::wrapped_to_raw_axis(b).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.
///
/// #328: matches via [`crate::axes::raw_matches_wrapped`] rather than exact
/// wrapped-name equality, so a RAW incoming name (`fast.length`) resolves
/// against the WRAPPED open/bound space exactly like an exact wrapped hit
/// (`sma_signal.fast.length`) would — the one shared suffix convention
/// `axes.rs` defines, reused rather than re-derived here. This is the
/// resolution mechanism, not an acceptance gate (contrast the CLI's own
/// `--axis` intake, which refuses a wrapped name before this ever runs on the
/// sweep-verb route); callers whose axes are not intake-filtered this cycle
/// (the walk-forward routes) keep accepting either form unchanged.
pub fn override_paths(
    axes: &[(String, Vec<Scalar>)],
    open_space: &[ParamSpec],
    signal: &Composite,
) -> Result<Vec<String>, String> {
    let bound = wrapped_bound_names(signal);
    let mut overrides = Vec::new();
    for (name, _) in axes {
        if open_space.iter().any(|p| crate::axes::raw_matches_wrapped(name, &p.name)) {
            continue;
        }
        match bound.iter().find(|b| crate::axes::raw_matches_wrapped(name, b)) {
            Some(b) => overrides.push(crate::axes::wrapped_to_raw_axis(b).to_string()),
            None => {
                return Err(format!(
                    "axis {name}: names no param of this blueprint (open or bound) — \
                     see `aura graph introspect --params <bp>`"
                ));
            }
        }
    }
    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");
    g.doc("rolling-extreme breakout latched into a long/short 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");
    g.doc("EMA deviation against a volatility band, latched into a mean-reversion 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");
    g.doc("prior high/low channel breaks latched into a directional signal");
    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;
    use crate::tap_plan::{bind_tap_plan, TapPlanError, TapSubscription};
    use crate::project::ProjectEnv;
    use std::path::{Path, PathBuf};

    #[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",
        )
        .expect("no cost model: nothing to refuse");
        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",
        )
        .expect("no cost model: nothing to refuse");
        // 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, TapPlan::record_all(), None)
            .expect("r_breakout example runs");
        let (via_carve, _) = run_signal_r(r_breakout_signal(Some(R_BREAKOUT_CHANNEL)), &[], RunData::Synthetic, 0, &env, TapPlan::record_all(), None)
            .expect("carved r_breakout signal runs");
        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, TapPlan::record_all(), None)
            .expect("r_meanrev example runs");
        let (via_carve, _) = run_signal_r(
            r_meanrev_signal(Some(R_MEANREV_WINDOW), Some(R_MEANREV_BAND_K)),
            &[],
            RunData::Synthetic,
            0,
            &env,
            TapPlan::record_all(),
            None,
        )
        .expect("carved r_meanrev signal runs");
        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).expect("test data resolves").0;
        let (_, window, pip) =
            resolve_run_data(&RunData::Synthetic, &env, &binding).expect("test data resolves");
        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",
            )
            .expect("GER40-keyed cost fixture resolves")
        };
        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"
        );
    }

    #[test]
    /// Review fix (#297 fork 3, item 1): the worker-channel seam a live sweep
    /// observes — `run_blueprint_member` given a cost model whose per-instrument
    /// map lacks the requested instrument refuses `Err(MemberFault::Bind(msg))`
    /// with `msg` byte-identical to `CostKnobFault::message()` (the
    /// `cost_knob_fault_message_is_byte_pinned` pin in `translate.rs`, one level
    /// up the call chain) — never a panic that `catch_unwind` would instead
    /// record as `MemberFault::Panic`. The fault fires before the harness is
    /// even bootstrapped, so `sources`/`window` are minimal placeholders.
    fn run_blueprint_member_propagates_the_cost_knob_fault_as_a_bind_message() {
        let env = Env::std();
        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 stop = StopRule::Vol { length: 3, k: 2.0 };
        let window = (Timestamp(0), Timestamp(0));
        let cost = [aura_research::CostSpec::Constant {
            cost_per_trade: aura_research::CostValue::PerInstrument(std::collections::BTreeMap::from(
                [("EURUSD".to_string(), 2.0)],
            )),
        }];
        let err = match run_blueprint_member(
            signal, &[], &[], Vec::new(), window, 0, 1.0, "topo", &env, stop, &binding, &cost, "GER40",
        ) {
            Err(e) => e,
            Ok(_) => panic!("GER40 is not in the map — the run must refuse, not succeed"),
        };
        let expected =
            crate::translate::CostKnobFault { knob: "cost_per_trade", instrument: "GER40".to_string() }
                .message();
        assert_eq!(err, MemberFault::Bind(expected));
    }

    /// #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).expect("test data resolves");
        let run = run_blueprint_member(
            reload(),
            &[],
            &space,
            sources,
            window,
            0,
            pip,
            "topo",
            &env,
            stop,
            &binding,
            &[],
            "GER40",
        )
        .expect("no cost model: nothing to refuse");
        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
        );
    }

    /// #338 write-side: `run_blueprint_member` given `StopRule::Fixed` actually
    /// bootstraps and runs the member (the `risk_executor`/`FixedStop` arm,
    /// end-to-end over a real synthetic realization) and stamps `stop_distance`
    /// into the manifest under the key `stop_rule_from_params`'s round-trip
    /// reads back — the `run_blueprint_member_stamps_the_vol_tf_stop_knobs`
    /// precedent above, `Fixed` edition.
    #[test]
    fn run_blueprint_member_stamps_the_fixed_stop_distance() {
        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("fixedstamp", reload().input_roles(), &BTreeMap::new())
            .expect("the price role resolves");
        let stop = StopRule::Fixed(10.0);
        let (sources, window, pip) =
            resolve_run_data(&RunData::Synthetic, &env, &binding).expect("test data resolves");
        let run = run_blueprint_member(
            reload(),
            &[],
            &space,
            sources,
            window,
            0,
            pip,
            "topo",
            &env,
            stop,
            &binding,
            &[],
            "GER40",
        )
        .expect("no cost model: nothing to refuse");
        assert!(
            run.manifest.params.contains(&("stop_distance".to_string(), Scalar::f64(10.0))),
            "manifest must stamp stop_distance: {:?}",
            run.manifest.params
        );
        assert!(
            !run.manifest.params.iter().any(|(k, _)| k == "stop_length" || k == "stop_k"),
            "a Fixed stop stamps no vol knobs: {:?}",
            run.manifest.params
        );
    }

    /// A fresh project root so a run's `runs/` lands in a temp dir.
    fn temp_project_env(name: &str) -> (Env, PathBuf) {
        let root = std::env::temp_dir()
            .join(format!("aura-member-plan-{name}-{}", std::process::id()));
        let _ = std::fs::remove_dir_all(&root);
        std::fs::create_dir_all(&root).expect("create temp project root");
        let env = Env::with_project(ProjectEnv {
            root: root.clone(),
            toml: Default::default(),
            commit: None,
            native: None,
        });
        (env, root)
    }

    /// The shipped `examples/r_sma.json` with one declared tap on node 0
    /// ("fast", SMA(length=2)) field 0 — the `tap_recording.rs` patch shape.
    /// `Composite` is `!Clone`, so each run loads it afresh.
    fn tapped_r_sma() -> Composite {
        let doc = include_str!("../../aura-cli/examples/r_sma.json");
        let mut v: serde_json::Value = serde_json::from_str(doc).expect("parse r_sma.json");
        v["blueprint"]["taps"] =
            serde_json::json!([{"name": "fast_tap", "from": {"node": 0, "field": 0}}]);
        let patched = serde_json::to_string(&v).expect("re-serialize");
        blueprint_from_json(&patched, &|t| std_vocabulary(t)).expect("tapped r_sma loads")
    }

    fn read_tap_json(root: &Path) -> serde_json::Value {
        let text = std::fs::read_to_string(
            root.join("runs").join("traces").join("sma_signal").join("fast_tap.json"),
        )
        .expect("persisted tap trace");
        serde_json::from_str(&text).expect("parse tap trace")
    }

    /// #283 acceptance: a fold plan's one-row summary and a live plan's
    /// collected stream both agree with the recorded series of the same
    /// blueprint (C1 — three identical runs, three drain policies), and a
    /// live-only plan persists nothing at all.
    #[test]
    fn fold_and_live_plans_agree_with_the_recorded_series() {
        // Run A: record (the charting question).
        let (env_a, root_a) = temp_project_env("record");
        run_signal_r(tapped_r_sma(), &[], RunData::Synthetic, 0, &env_a, TapPlan::record_all(), None)
            .expect("run A records");
        let series = read_tap_json(&root_a);
        let ts: Vec<i64> =
            series["ts"].as_array().unwrap().iter().map(|v| v.as_i64().unwrap()).collect();
        let vals: Vec<f64> =
            series["columns"][0].as_array().unwrap().iter().map(|v| v.as_f64().unwrap()).collect();
        assert!(!vals.is_empty(), "the recorded series is the reference");

        // Run B: fold to mean (the aggregate question, same blueprint).
        let (env_b, root_b) = temp_project_env("fold");
        let mut plan_b = TapPlan::record_all();
        plan_b.subscribe("fast_tap", TapSubscription::named("mean"));
        run_signal_r(tapped_r_sma(), &[], RunData::Synthetic, 0, &env_b, plan_b, None)
            .expect("run B folds");
        let row = read_tap_json(&root_b);
        let mean = vals.iter().sum::<f64>() / vals.len() as f64; // sequential — FoldState's order
        assert_eq!(row["ts"], serde_json::json!([*ts.last().unwrap()]));
        assert_eq!(row["columns"], serde_json::json!([[mean]]), "fold row == slice mean, bit-exact");

        // Run C: live only (nothing persisted; the closure sees the series).
        let (env_c, root_c) = temp_project_env("live");
        let (live_tx, live_rx) = std::sync::mpsc::channel();
        let mut plan_c = TapPlan::empty();
        plan_c.subscribe(
            "fast_tap",
            TapSubscription::live(move |ts, cell| {
                let _ = live_tx.send((ts.0, cell.f64()));
            }),
        );
        run_signal_r(tapped_r_sma(), &[], RunData::Synthetic, 0, &env_c, plan_c, None)
            .expect("run C runs live-only");
        let got: Vec<(i64, f64)> = live_rx.try_iter().collect();
        let want: Vec<(i64, f64)> = ts.iter().copied().zip(vals.iter().copied()).collect();
        assert_eq!(got, want, "the live closure saw exactly the recorded series (C1)");
        assert!(!root_c.join("runs").exists(), "a live-only plan persists nothing");
    }

    /// Plan refusals are typed and fire before any store I/O.
    #[test]
    fn bind_tap_plan_refuses_unknown_tap_and_unknown_label_before_the_store() {
        let (env, root) = temp_project_env("refusals");

        // Unknown tap name.
        let mut flat = tapped_r_sma().compile_with_params(&[]).expect("tapped r_sma compiles");
        let mut plan = TapPlan::record_all();
        plan.subscribe("no_such_tap", TapSubscription::named("mean"));
        // `BoundTaps` (the `Ok` side) carries no `Debug`, so `expect_err` is
        // unavailable here — match explicitly instead (same refusal intent).
        let err = match bind_tap_plan(&mut flat, plan, &env, "sma_signal") {
            Err(e) => e,
            Ok(_) => panic!("unknown tap must be refused"),
        };
        assert!(
            matches!(err, TapPlanError::UnknownTap { ref name, .. } if name == "no_such_tap"),
            "{err:?}"
        );
        let msg = err.to_string();
        assert!(
            msg.contains("fast_tap"),
            "unknown-tap refusal enumerates the declared taps (fixture declares only fast_tap): {msg}"
        );

        // Unknown label — the refusal enumerates the roster.
        let mut flat2 = tapped_r_sma().compile_with_params(&[]).expect("tapped r_sma compiles");
        let mut plan2 = TapPlan::record_all();
        plan2.subscribe("fast_tap", TapSubscription::named("p95"));
        let err2 = match bind_tap_plan(&mut flat2, plan2, &env, "sma_signal") {
            Err(e) => e,
            Ok(_) => panic!("unknown label must be refused"),
        };
        let msg = err2.to_string();
        assert!(msg.starts_with("unknown fold 'p95'") && msg.contains("mean"), "{msg}");

        assert!(!root.join("runs").exists(), "refusals fire before any store write");
    }

    /// C1: two identical record-all runs produce byte-identical tap files.
    #[test]
    fn two_identical_record_runs_produce_byte_identical_tap_files() {
        let (env_a, root_a) = temp_project_env("det-a");
        let (env_b, root_b) = temp_project_env("det-b");
        run_signal_r(tapped_r_sma(), &[], RunData::Synthetic, 0, &env_a, TapPlan::record_all(), None)
            .expect("run a records");
        run_signal_r(tapped_r_sma(), &[], RunData::Synthetic, 0, &env_b, TapPlan::record_all(), None)
            .expect("run b records");
        let a = std::fs::read(
            root_a.join("runs").join("traces").join("sma_signal").join("fast_tap.json"),
        )
        .expect("run a trace");
        let b = std::fs::read(
            root_b.join("runs").join("traces").join("sma_signal").join("fast_tap.json"),
        )
        .expect("run b trace");
        assert_eq!(a, b, "same input, same bytes (the index carries env provenance; the tap file must not)");
    }

    /// #297: a refusal returns through the Result seam — the embedding
    /// process survives to read it (the #283 fieldtest scenario that used
    /// to kill the host).
    #[test]
    fn run_signal_r_refuses_in_process_instead_of_killing_the_host() {
        let (env, _root) = temp_project_env("in-process-refusal");
        let mut plan = TapPlan::record_all();
        plan.subscribe("fast_tap", TapSubscription::named("median"));
        let err = run_signal_r(tapped_r_sma(), &[], RunData::Synthetic, 0, &env, plan, None)
            .unwrap_err();
        assert_eq!(err.exit_code, 2);
        assert!(err.message.contains("unknown fold"), "{}", err.message);
    }

    /// The #297 promise at the exposes-neither seam (fieldtest
    /// milestone-safe-to-embed, finding B1): a blueprint exposing NEITHER a
    /// `bias` output NOR a declared tap is not runnable on this leg, and the
    /// library must say so as a returned `RunnerError` — C14 class 2
    /// (argv-named content, the CLI guard's own prose family) — never via
    /// `wrap_r`'s wiring `.expect` (`UnknownOutPort { name: "bias" }`,
    /// member.rs:443) panicking and killing the embedding host. The CLI's
    /// shell-side pre-validation (`exec_blueprint_leg`) must not be the only
    /// guard on the seam the milestone promises is kill-free.
    #[test]
    fn run_signal_r_refuses_an_exposes_neither_blueprint_in_process() {
        let (env, _root) = temp_project_env("exposes-neither");
        // Minimal exposes-neither signal: one role, one node, no exposed
        // output, no declared tap.
        let mut g = GraphBuilder::new("blind_signal");
        let s = g.add(Sub::builder());
        let price = g.source_role("price", ScalarKind::F64);
        g.feed(price, [s.input("lhs"), s.input("rhs")]);
        let blind = g.build().expect("an exposes-neither composite still builds");
        let err = run_signal_r(blind, &[], RunData::Synthetic, 0, &env, TapPlan::empty(), None)
            .unwrap_err();
        assert_eq!(err.exit_code, 2, "argv-named content (C14): {}", err.message);
        assert!(err.message.contains("bias"), "{}", err.message);
        assert!(err.message.contains("exposes neither"), "{}", err.message);
    }
}