Aura/crates/aura-cli/src/campaign_run.rs

//! `aura campaign run` — the driver that turns a stored campaign document into
//! a realized run-set (#198). The execution *semantics* (preflight, cell loop,
//! stage sequencing, selection, registry writes) live in `aura-campaign`; this
//! module owns what is CLI-specific: target resolution (file sugar vs content
//! id), the project + referential gates, the [`MemberRunner`] implementation
//! over the shipped loaded-blueprint machinery (`wrap_r` reduce-mode member
//! runs via `run_blueprint_member` + `M1FieldSource` windowed real-data
//! binding), fault prose (`exec_fault_prose` lives HERE, beside its consumer,
//! not in `research_docs.rs` — it phrases `aura_campaign` types the doc-tier
//! module deliberately does not import), and stdout/stderr emission.
//!
//! Root items (`blueprint_axis_probe`, `run_blueprint_member`,
//! `family_member_line`) are reached via `crate::` — the `graph_construct`
//! submodule idiom: main.rs is the crate root, so its private items are
//! visible to child modules without promotion.

use std::path::{Path, PathBuf};
use std::sync::{mpsc, Arc};

use aura_campaign::{CampaignOutcome, CellSpec, ExecFault, MemberFault, MemberRunner};
use aura_core::{Cell, ParamSpec, Scalar, ScalarKind, Timestamp};
use aura_engine::{
    blueprint_from_json, f64_field, summarize, summarize_r, ColumnarTrace, FamilySelection,
    RunReport,
};
use aura_ingest::{instrument_geometry, unix_ms_to_epoch_ns, M1Field, M1FieldSource};
use aura_registry::{CampaignRunRecord, WriteKind};
use aura_research::{
    campaign_to_json, content_id_of, parse_campaign, parse_process, validate_campaign,
    validate_process, CampaignDoc, DocRef,
};

use crate::project::Env;
use crate::research_docs::{
    doc_error_prose, doc_fault_prose, fault_block, parse_valid_campaign, ref_fault_prose,
};

/// A bare store address: exactly 64 lowercase hex chars (the content-id key
/// shape). `pub(crate)`: `graph_construct`'s `--params <FILE|ID>` resolution
/// (#196) shares this exact predicate rather than re-inlining it, so the two
/// FILE-or-id surfaces cannot drift apart on the id shape a second time.
pub(crate) fn is_content_id(s: &str) -> bool {
    s.len() == 64 && s.bytes().all(|b| matches!(b, b'0'..=b'9' | b'a'..=b'f'))
}

/// Phrase an [`ExecFault`] for stderr — Debug-leak-free, path-addressed
/// (stage index + block id), the `doc_fault_prose`/`ref_fault_prose` register.
pub(crate) fn exec_fault_prose(f: &ExecFault) -> String {
    match f {
        ExecFault::PipelineShape { detail } => {
            format!("process pipeline is not executable: {detail}")
        }
        ExecFault::UnrankableMetric { stage, metric } => format!(
            "process stage {stage}: metric \"{metric}\" is not rankable (winner \
             selection needs one of the registry's rankable metrics; see: aura \
             process introspect --metrics)"
        ),
        ExecFault::GateMetricNotPerMember { stage, metric } => format!(
            "process stage {stage}: gate metric \"{metric}\" is not a per-member \
             scalar (selection annotations cannot gate members; see: aura process \
             introspect --metrics)"
        ),
        ExecFault::PlateauInWalkForward { stage } => format!(
            "process stage {stage}: walk_forward cannot use a plateau select (a \
             gated survivor subset has no parameter lattice)"
        ),
        ExecFault::DeflatePlateauConflict { stage } => format!(
            "process stage {stage}: sweep deflate: true composes only with select \"argmax\""
        ),
        ExecFault::GeneralizeNeedsInstruments { available } => format!(
            "campaign has {available} instrument(s); std::generalize needs at least 2"
        ),
        ExecFault::GeneralizeNonRMetric { metric } => format!(
            "std::generalize metric \"{metric}\" is not in the rankable R-expectancy \
             family (expectancy_r, net_expectancy_r, sqn, sqn_normalized generalize; \
             see: aura process introspect --metrics)"
        ),
        ExecFault::ZeroBootstrapParam { stage, field } => format!(
            "process stage {stage}: monte_carlo {field} must be > 0"
        ),
        ExecFault::Window { stage, detail } => format!(
            "process stage {stage}: walk_forward windows do not fit the campaign \
             window: {detail}"
        ),
        ExecFault::Member(MemberFault::NoData { instrument, window_ms }) => format!(
            "no data for instrument {instrument} in window [{}, {}] (epoch-ms)",
            window_ms.0, window_ms.1
        ),
        ExecFault::Member(MemberFault::Bind(detail)) => {
            format!("a member failed to bind: {detail}")
        }
        ExecFault::Member(MemberFault::Run(detail)) => {
            format!("a member failed to run: {detail}")
        }
        ExecFault::Registry(e) => e.to_string(),
    }
}

/// stdout wire form of one selection-bearing stage. Field order is the wire
/// contract (serde derives declaration order).
#[derive(serde::Serialize)]
struct SelectionReportLine<'a> {
    selection_report: SelectionReportBody<'a>,
}

#[derive(serde::Serialize)]
struct SelectionReportBody<'a> {
    family_id: &'a str,
    stage: usize,
    block: &'a str,
    winner_ordinal: usize,
    params: &'a [(String, Scalar)],
    selection: &'a FamilySelection,
}

/// The always-on final line: the stored realization record under one key.
#[derive(serde::Serialize)]
struct CampaignRunLine<'a> {
    campaign_run: &'a CampaignRunRecord,
}

/// Suffix-join each raw campaign axis onto exactly one wrapped param (wrapped
/// == raw, or wrapped ends with ".{raw}" — the established suffix-match
/// pattern), then require every wrapped slot to be covered. Pure and
/// independent of the env/data seam so the three fault arms are unit-testable
/// without a project, a store, or a loaded blueprint.
fn bind_axes(
    space: &[ParamSpec],
    strategy_id: &str,
    params: &[(String, Scalar)],
) -> Result<Vec<Cell>, MemberFault> {
    let mut slots: Vec<Option<Scalar>> = vec![None; space.len()];
    for (raw, value) in params {
        let hits: Vec<usize> = space
            .iter()
            .enumerate()
            .filter(|(_, p)| p.name == *raw || p.name.ends_with(&format!(".{raw}")))
            .map(|(i, _)| i)
            .collect();
        match hits.as_slice() {
            [i] => slots[*i] = Some(*value),
            [] => {
                return Err(MemberFault::Bind(format!(
                    "axis \"{raw}\" matches no open param of the wrapped strategy {strategy_id}"
                )));
            }
            _ => {
                let names: Vec<&str> = hits.iter().map(|&i| space[i].name.as_str()).collect();
                return Err(MemberFault::Bind(format!(
                    "axis \"{raw}\" is ambiguous in the wrapped param space of \
                     strategy {strategy_id}: matches {}",
                    names.join(", ")
                )));
            }
        }
    }
    let mut point: Vec<Cell> = Vec::with_capacity(space.len());
    for (spec, slot) in space.iter().zip(&slots) {
        match slot {
            Some(v) => point.push(v.cell()),
            None => {
                // Speak the RAW campaign-axis namespace (the doc's own): the
                // wrapped space prefixes the signal's params with one node
                // segment, so the raw path is everything after it (#203).
                let raw =
                    spec.name.split_once('.').map(|(_, rest)| rest).unwrap_or(&spec.name);
                return Err(MemberFault::Bind(format!(
                    "open param \"{raw}\" of strategy {strategy_id} is bound by no \
                     campaign axis (wrapped path: {})",
                    spec.name
                )));
            }
        }
    }
    Ok(point)
}

/// The CLI's harness/data binding seam for `aura_campaign::execute`: members
/// run through the shipped loaded-blueprint machinery (`wrap_r` reduce-mode
/// via `crate::run_blueprint_member`) over windowed real M1 close bars
/// (`M1FieldSource::open_window` — the ms→ns crossing happens at exactly this
/// seam, via `unix_ms_to_epoch_ns`). All refusals are member faults for the
/// library to surface; never a process exit inside a sweep worker.
struct CliMemberRunner<'a> {
    env: &'a Env,
    server: Arc<data_server::DataServer>,
}

impl MemberRunner for CliMemberRunner<'_> {
    fn run_member(
        &self,
        cell: &CellSpec,
        params: &[(String, Scalar)],
        window_ms: (i64, i64),
    ) -> Result<RunReport, MemberFault> {
        // The wrapped axis namespace — the SAME probe the sweep verbs resolve
        // against (single-sourced in main.rs; identical `false, true, None` wrap).
        let space = crate::blueprint_axis_probe(&cell.blueprint_json, self.env).param_space();
        let point = bind_axes(&space, &cell.strategy_id, params)?;

        // Real windowed data — geometry BEFORE bar data (the shipped pre-data
        // refusal order of `open_real_source`), both as member faults.
        let geo = instrument_geometry(&self.server, &cell.instrument).ok_or_else(|| {
            MemberFault::Run(format!(
                "no recorded geometry for symbol '{}' at {} — refusing to run a \
                 real instrument with a guessed pip",
                cell.instrument,
                self.env.data_path()
            ))
        })?;
        let from = unix_ms_to_epoch_ns(window_ms.0);
        let to = unix_ms_to_epoch_ns(window_ms.1);
        let source = match M1FieldSource::open_window(
            &self.server,
            &cell.instrument,
            Some(from),
            Some(to),
            M1Field::Close,
        ) {
            Some(s) => s,
            // No archived file overlaps the window at all.
            None => {
                return Err(MemberFault::NoData {
                    instrument: cell.instrument.clone(),
                    window_ms,
                });
            }
        };
        // A window that overlaps a file but holds zero matching bars yields a
        // source whose first peek is None (open_window's documented contract)
        // — the same no-data condition.
        if aura_engine::Source::peek(&source).is_none() {
            return Err(MemberFault::NoData {
                instrument: cell.instrument.clone(),
                window_ms,
            });
        }

        // The shipped member recipe (reload — a Composite is !Clone — wrap,
        // bind, run, summarize): `run_blueprint_member` verbatim. Seed 0
        // (seed-free real-data runs), topology_hash = the strategy's content
        // id, project provenance stamped inside the helper.
        let signal = blueprint_from_json(&cell.blueprint_json, &|t| self.env.resolve(t))
            .expect("stored blueprint passed the referential gate; reload is infallible");
        let mut report = crate::run_blueprint_member(
            signal,
            &point,
            &space,
            vec![Box::new(source)],
            (from, to),
            0,
            geo.pip_size,
            &cell.strategy_id,
            self.env,
        );
        report.manifest.instrument = Some(cell.instrument.clone());
        Ok(report)
    }
}

/// `aura campaign run <target>`: resolve, gate, execute, emit. Every `Err`
/// is a refusal `campaign_cmd` renders as `aura: {msg}` + exit 1.
pub(crate) fn run_campaign(target: &str, env: &Env) -> Result<(), String> {
    // Project gate FIRST: nothing (not even the file-sugar registration)
    // touches a store outside a project.
    if env.provenance().is_none() {
        let cwd = std::env::current_dir()
            .map(|d| d.display().to_string())
            .unwrap_or_default();
        return Err(format!(
            "campaign run needs a project: strategies resolve against the project \
             store and vocabulary (no Aura.toml found up from {cwd})"
        ));
    }
    let registry = env.registry();

    // Target resolution: a readable file is register-then-run sugar; a bare
    // 64-hex token addresses the store directly; anything else refuses naming
    // both readings. A CLI arg is ephemeral input, so a leading `content:`
    // display prefix is tolerated here (#194) — doc ref FIELDS stay bare-only
    // (canonical-form byte stability).
    let target = target.strip_prefix("content:").filter(|t| is_content_id(t)).unwrap_or(target);
    let campaign_id = if Path::new(target).is_file() {
        let doc = parse_valid_campaign(&PathBuf::from(target))?;
        registry
            .put_campaign(&campaign_to_json(&doc))
            .map_err(|e| e.to_string())?
    } else if is_content_id(target) {
        target.to_string()
    } else {
        return Err(format!(
            "'{target}' is neither a readable .json file nor a 64-hex content id"
        ));
    };

    // One uniform path from here: fetch the stored canonical bytes by id (so
    // file addressing and id addressing produce the same realization by
    // construction) and re-run the intrinsic tier on them.
    let campaign_text = registry
        .get_campaign(&campaign_id)
        .map_err(|e| e.to_string())?
        .ok_or_else(|| format!("no campaign {campaign_id} in the project store"))?;
    let campaign = parse_campaign(&campaign_text)
        .map_err(|e| doc_error_prose("stored campaign document", &e))?;
    let faults = validate_campaign(&campaign);
    if !faults.is_empty() {
        return Err(fault_block(
            "campaign document invalid:",
            faults.iter().map(doc_fault_prose).collect(),
        ));
    }

    // Referential gate: zero faults or refuse (the campaign-validate seam).
    let resolve = |t: &str| env.resolve(t);
    let ref_faults = registry
        .validate_campaign_refs(&campaign, &resolve)
        .map_err(|e| e.to_string())?;
    if !ref_faults.is_empty() {
        return Err(fault_block(
            "campaign references do not resolve:",
            ref_faults.iter().map(ref_fault_prose).collect(),
        ));
    }

    // Process fetch + intrinsic validation (stored text, not a file path —
    // parse_valid_process is file-based, so its constituents run here).
    let DocRef::ContentId(process_id) = &campaign.process.r#ref else {
        // validate_campaign already refuses identity process refs; defensive.
        return Err("process.ref: a process is referenced by content id in this version".into());
    };
    let process_text = registry
        .get_process(process_id)
        .map_err(|e| e.to_string())?
        .ok_or_else(|| format!("no process {process_id} in the project store"))?;
    let process = parse_process(&process_text)
        .map_err(|e| doc_error_prose("stored process document", &e))?;
    let process_faults = validate_process(&process);
    if !process_faults.is_empty() {
        return Err(fault_block(
            "process document invalid:",
            process_faults.iter().map(doc_fault_prose).collect(),
        ));
    }

    // Strategies: canonical bytes from the store, index-aligned with the doc.
    // The recorded id is the content id of those bytes (== the ref id for
    // content refs; computed for identity refs).
    let mut strategies: Vec<(String, String)> = Vec::with_capacity(campaign.strategies.len());
    for entry in &campaign.strategies {
        let canonical = match &entry.r#ref {
            DocRef::ContentId(id) => registry
                .get_blueprint(id)
                .map_err(|e| e.to_string())?
                .ok_or_else(|| format!("strategy {id} not found in the blueprint store"))?,
            DocRef::IdentityId(id) => registry
                .find_blueprint_by_identity(id, &resolve)
                .map_err(|e| e.to_string())?
                .ok_or_else(|| format!("identity id {id} matches no stored blueprint"))?,
        };
        strategies.push((content_id_of(&canonical), canonical));
    }

    let runner = CliMemberRunner {
        env,
        server: Arc::new(data_server::DataServer::new(env.data_path())),
    };
    let outcome = aura_campaign::execute(
        &campaign_id,
        &campaign,
        &process,
        &strategies,
        &runner,
        &registry,
    )
    .map_err(|f| exec_fault_prose(&f))?;

    // Zero-survivor stderr notes (exit stays 0 — a null result is a valid
    // research result, #198 decision 8). Addressed by the fields the record
    // already carries (strategy/instrument/window_ms), not by re-deriving a
    // doc-order ordinal from the loop-nesting the executor happens to use
    // today — that duplicated invariant would silently drift if the executor
    // ever reorders its cell loop.
    for cell in &outcome.record.cells {
        for (stage_ix, st) in cell.stages.iter().enumerate() {
            if matches!(&st.survivor_ordinals, Some(v) if v.is_empty()) {
                eprintln!(
                    "aura: cell {}/{}/[{}, {}]: gate at stage {stage_ix} left no \
                     survivors; cell realization truncated",
                    cell.strategy, cell.instrument, cell.window_ms.0, cell.window_ms.1
                );
            }
        }
    }

    // Emission: emit-gated family/selection lines per cell, then the
    // always-on final record line. Matched by NAME against the two closed-
    // vocabulary terms (self-evident at the call site, unlike positional
    // indexing into the vocab slice); the debug_assert keeps the names
    // honest against `aura_research::emit_vocabulary()` so a #190
    // rename/extend of the closed set fails loudly here instead of silently
    // misrouting emission.
    debug_assert!(
        aura_research::emit_vocabulary().contains(&"family_table")
            && aura_research::emit_vocabulary().contains(&"selection_report"),
        "emit_vocabulary drifted from the names campaign_run matches by"
    );
    let emit_family = campaign.presentation.emit.iter().any(|e| e == "family_table");
    let emit_selection = campaign.presentation.emit.iter().any(|e| e == "selection_report");
    for cell_out in &outcome.cells {
        if emit_family {
            for fam in &cell_out.families {
                for report in &fam.reports {
                    println!("{}", crate::family_member_line(&fam.family_id, report));
                }
            }
        }
        if emit_selection {
            for sel in &cell_out.selections {
                let line = SelectionReportLine {
                    selection_report: SelectionReportBody {
                        family_id: &sel.family_id,
                        stage: sel.stage,
                        block: sel.block,
                        winner_ordinal: sel.winner_ordinal,
                        params: &sel.params,
                        selection: &sel.selection,
                    },
                };
                println!(
                    "{}",
                    serde_json::to_string(&line).expect("selection report serializes")
                );
            }
        }
    }
    println!(
        "{}",
        serde_json::to_string(&CampaignRunLine { campaign_run: &outcome.record })
            .expect("campaign run record serializes")
    );

    // Trace persistence runs LAST, after the always-on record line (0109):
    // stdout stays data-pure (the record line is the wire contract) and the
    // trace surface is a stderr concern; the realization record is already
    // stored by `execute`, so a trace failure exits 1 without un-recording it.
    if let Some(trace_name) = &outcome.record.trace_name {
        persist_campaign_traces(
            trace_name,
            &campaign.presentation.persist_taps,
            &outcome,
            &campaign,
            &strategies,
            &runner.server,
            env,
        )?;
    }
    Ok(())
}

/// Which drained wrap-convention channel a requested tap persists from on a
/// campaign trace re-run. The routing mirrors `persist_traces_r` (main.rs):
/// equity <- the SimBroker equity recorder (tx_eq), exposure <- the Bias
/// recorder (tx_ex), r_equity <- the cum_realized_r + unrealized_r recorder
/// (tx_req). `net_r_equity` has no channel here: the campaign runner wires
/// no cost leg (`wrap_r(.., cost: None)`), so a net curve is not produced
/// by this run's configuration.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum TapChannel {
    Equity,
    Exposure,
    REquity,
}

/// Route one requested tap of the closed vocabulary to its drained channel;
/// `None` marks a tap this run cannot produce (the caller skips it loudly).
/// An unknown name cannot reach here (`validate_campaign` refuses it) and
/// maps to `None` all the same rather than guessing.
fn tap_channel(tap: &str) -> Option<TapChannel> {
    // The emit_vocabulary debug_assert's twin: every vocabulary tap must be
    // either routable here or the one known cost-only tap — a future fifth
    // vocabulary entry fails loudly instead of silently skipping with the
    // hardcoded needs-a-cost-run message.
    debug_assert!(
        aura_research::tap_vocabulary()
            .iter()
            .all(|t| *t == "net_r_equity"
                || matches!(*t, "equity" | "exposure" | "r_equity")),
        "tap_vocabulary drifted from the channels campaign_run routes"
    );
    match tap {
        "equity" => Some(TapChannel::Equity),
        "exposure" => Some(TapChannel::Exposure),
        "r_equity" => Some(TapChannel::REquity),
        _ => None,
    }
}

/// The content-derived on-disk key of one campaign cell under
/// `traces/<trace_name>/` — the `member_key` discipline (content, never a
/// runtime ordinal): strategy content-id prefix + instrument +
/// doc-positional window ordinal, sanitized to one portable path component.
fn campaign_cell_key(strategy: &str, instrument: &str, window_ordinal: usize) -> String {
    let strategy8 = strategy.get(..8).unwrap_or(strategy);
    crate::sanitize_component(&format!("{strategy8}-{instrument}-w{window_ordinal}"))
}

/// Persist the requested `persist_taps` for every nominee cell under
/// `traces/<trace_name>/<cell_key>/` (0109, #201 d5): re-run each nominee
/// once in non-reduce trace mode over its own recorded `manifest.window`,
/// assert the re-run METRICS equal the recorded nominee metrics (the C1
/// drift alarm — manifest fields are fresh-context and not compared), and
/// write the requested-AND-producible taps through the sweep verbs'
/// member-dir mechanism: a slash-joined `"{name}/{key}"` handed to
/// `TraceStore::write` (the `persist_traces_r(&format!("{name}/{key}"), ..)`
/// layout — the fn itself is not reused: it writes a fixed tap set and
/// process-exits on error). The written index manifest is the RECORDED
/// nominee's — the trace documents that run's provenance, not a re-derived
/// fresh-context one. Loud stderr per no-nominee cell and ONCE per
/// unproducible requested tap (producibility is run-configuration-level,
/// not per-cell); one summary line at the end. Every `Err` is a refusal
/// `campaign_cmd` renders as `aura: {msg}` + exit 1.
fn persist_campaign_traces(
    trace_name: &str,
    taps: &[String],
    outcome: &CampaignOutcome,
    campaign: &CampaignDoc,
    strategies: &[(String, String)],
    server: &Arc<data_server::DataServer>,
    env: &Env,
) -> Result<(), String> {
    let store = env.trace_store();
    store
        .ensure_name_free(trace_name, WriteKind::Family)
        .map_err(|e| e.to_string())?;

    // Requested ∩ producible, in request order. When nothing producible was
    // requested (net_r_equity only), no member dir is written and the
    // summary honestly reports 0 tap(s).
    let mut routed: Vec<(&str, TapChannel)> = Vec::new();
    for tap in taps {
        match tap_channel(tap) {
            Some(ch) => routed.push((tap.as_str(), ch)),
            None => eprintln!(
                "aura: tap \"{tap}\" is not produced by this run (needs a cost run); skipped"
            ),
        }
    }

    // `outcome.cells` is index-aligned with `outcome.record.cells` by
    // construction: `execute` pushes both in the same cell-loop iteration.
    let mut persisted_cells = 0usize;
    for (cell_rec, cell_out) in outcome.record.cells.iter().zip(&outcome.cells) {
        let Some((_, nominee_report)) = &cell_out.nominee else {
            eprintln!(
                "aura: cell {}/{}/[{}, {}]: no nominee; no traces persisted",
                cell_rec.strategy, cell_rec.instrument, cell_rec.window_ms.0, cell_rec.window_ms.1
            );
            continue;
        };
        if routed.is_empty() {
            continue;
        }

        // The cell key is doc-derived: the window ordinal is the POSITION
        // of the cell's window in campaign.data.windows (mirroring the
        // family-name convention), never a loop counter re-derived here.
        let window_ordinal = campaign
            .data
            .windows
            .iter()
            .position(|w| (w.from_ms, w.to_ms) == cell_rec.window_ms)
            .ok_or_else(|| {
                format!(
                    "cell window [{}, {}] is not one of campaign.data.windows",
                    cell_rec.window_ms.0, cell_rec.window_ms.1
                )
            })?;
        let cell_key =
            campaign_cell_key(&cell_rec.strategy, &cell_rec.instrument, window_ordinal);

        // The cell's blueprint: the run's own index-aligned resolution, by
        // content id (exactly the bytes the executor's members ran).
        let (_, blueprint_json) = strategies
            .iter()
            .find(|(id, _)| *id == cell_rec.strategy)
            .ok_or_else(|| {
                format!(
                    "cell strategy {} is not among the run's resolved strategies",
                    cell_rec.strategy
                )
            })?;

        // Re-run the nominee, non-reduce: the SAME member the executor ran
        // (same wrapped space, same params, same window, seed-free real
        // data), mirroring `CliMemberRunner::run_member` with the reduce
        // fold off so the per-cycle tap streams exist. The window is the
        // nominee report's own `manifest.window` — already epoch-ns
        // (`run_blueprint_member` stamped the post-seam bounds), so no
        // second ms->ns crossing here.
        let space = crate::blueprint_axis_probe(blueprint_json, env).param_space();
        let point = crate::point_from_params(&space, &nominee_report.manifest.params);
        let (from, to) = nominee_report.manifest.window;
        let no_data = || {
            format!(
                "no data for instrument {} in the nominee window [{}, {}] (epoch-ns)",
                cell_rec.instrument, from.0, to.0
            )
        };
        let source = M1FieldSource::open_window(
            server,
            &cell_rec.instrument,
            Some(from),
            Some(to),
            M1Field::Close,
        )
        .ok_or_else(no_data)?;
        if aura_engine::Source::peek(&source).is_none() {
            return Err(no_data());
        }
        let signal = blueprint_from_json(blueprint_json, &|t| env.resolve(t))
            .expect("stored blueprint passed the referential gate; reload is infallible");
        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 mut h = crate::wrap_r(signal, tx_eq, tx_ex, tx_r, tx_req, false, false, None)
            .bootstrap_with_cells(&point)
            .expect("the nominee's point re-bootstraps (it already ran this realization)");
        let sources: Vec<Box<dyn aura_engine::Source>> = vec![Box::new(source)];
        h.run(sources);
        // Drain ALL FOUR channels (`run_signal_r` leaves req undrained; the
        // trace path must not): eq/ex/req feed the taps, r feeds the metrics.
        let eq_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_eq.try_iter().collect();
        let ex_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_ex.try_iter().collect();
        let r_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_r.try_iter().collect();
        let req_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_req.try_iter().collect();

        // The C1 drift alarm: metrics equality against the recorded
        // nominee. The reduce-mode fold shares its arithmetic with this
        // non-reduce reduction (SeriesFold via `summarize`; GatedRecorder
        // emits exactly the rows `summarize_r`'s ledger reads), so equality
        // is bit-exact.
        let mut rerun_metrics = summarize(&f64_field(&eq_rows, 0), &f64_field(&ex_rows, 0));
        rerun_metrics.r = Some(summarize_r(&r_rows, &[]));
        if rerun_metrics != nominee_report.metrics {
            return Err(format!(
                "trace re-run diverged from the recorded nominee (C1 violation): cell \
                 {cell_key} of trace {trace_name} does not reproduce its recorded \
                 metrics; refusing to persist a silently-wrong trace"
            ));
        }

        let traces: Vec<ColumnarTrace> = routed
            .iter()
            .map(|&(tap, ch)| {
                let rows: &[(Timestamp, Vec<Scalar>)] = match ch {
                    TapChannel::Equity => &eq_rows,
                    TapChannel::Exposure => &ex_rows,
                    TapChannel::REquity => &req_rows,
                };
                ColumnarTrace::from_rows(tap, &[ScalarKind::F64], rows)
            })
            .collect();
        store
            .write(&format!("{trace_name}/{cell_key}"), &nominee_report.manifest, &traces)
            .map_err(|e| e.to_string())?;
        persisted_cells += 1;
    }

    eprintln!(
        "aura: traces persisted: {trace_name} ({} tap(s) x {persisted_cells} cell(s))",
        routed.len()
    );
    Ok(())
}

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

    fn spec(name: &str) -> ParamSpec {
        ParamSpec { name: name.to_string(), kind: ScalarKind::I64 }
    }

    #[test]
    /// #197 (fieldtest 0107 F9): the executor's preflight refusal for a
    /// non-rankable selection metric points the author at the verb that
    /// enumerates the applicable roster, so the exists-but-not-rankable split
    /// (a metric emitted in every member's block yet refused for winner
    /// selection) is resolvable without trial and error. The intrinsic message
    /// is unchanged; only a discovery pointer is appended.
    fn unrankable_metric_prose_points_at_the_metrics_introspection() {
        let prose = exec_fault_prose(&ExecFault::UnrankableMetric {
            stage: 0,
            metric: "profit_factor".into(),
        });
        assert!(prose.contains("is not rankable"), "intrinsic message kept: {prose}");
        assert!(
            prose.contains("aura process introspect --metrics"),
            "the refusal must point at the enumeration verb: {prose}"
        );
    }

    #[test]
    /// #197: the twin refusal — a gate predicate over a non-per-member scalar
    /// (a selection annotation) — likewise points at the enumeration verb.
    fn gate_metric_not_per_member_prose_points_at_the_metrics_introspection() {
        let prose = exec_fault_prose(&ExecFault::GateMetricNotPerMember {
            stage: 1,
            metric: "deflated_score".into(),
        });
        assert!(prose.contains("is not a per-member"), "intrinsic message kept: {prose}");
        assert!(
            prose.contains("aura process introspect --metrics"),
            "the refusal must point at the enumeration verb: {prose}"
        );
    }

    #[test]
    /// generalize's static instrument-arity refusal is Debug-free and
    /// names the available count plus the required floor.
    fn generalize_needs_instruments_prose_names_the_shortfall() {
        let prose = exec_fault_prose(&ExecFault::GeneralizeNeedsInstruments { available: 1 });
        assert_eq!(prose, "campaign has 1 instrument(s); std::generalize needs at least 2");
        assert!(!prose.contains("GeneralizeNeedsInstruments"), "Debug leak: {prose}");
    }

    #[test]
    /// #207 (fieldtest 0108 F8): generalize's refusal names the REAL rule — the
    /// rankable R-expectancy family that generalizes — and points at the roster
    /// verb, instead of the old "pip metrics do not" mislabel (which was wrong
    /// for R-denominated-but-unranked names like max_r_drawdown). The four
    /// accepted names are enumerated in the prose as a fail-safe hint; the
    /// durable source is the `(see: …)` roster pointer.
    fn generalize_non_r_metric_prose_names_the_metric() {
        let prose = exec_fault_prose(&ExecFault::GeneralizeNonRMetric {
            metric: "total_pips".into(),
        });
        assert_eq!(
            prose,
            "std::generalize metric \"total_pips\" is not in the rankable R-expectancy \
             family (expectancy_r, net_expectancy_r, sqn, sqn_normalized generalize; \
             see: aura process introspect --metrics)"
        );
        assert!(!prose.contains("GeneralizeNonRMetric"), "Debug leak: {prose}");
    }

    #[test]
    /// the zero-bootstrap-param refusal is path-addressed (stage index)
    /// and names the offending field.
    fn zero_bootstrap_param_prose_names_stage_and_field() {
        let prose = exec_fault_prose(&ExecFault::ZeroBootstrapParam {
            stage: 3,
            field: "block_len",
        });
        assert_eq!(prose, "process stage 3: monte_carlo block_len must be > 0");
        assert!(!prose.contains("ZeroBootstrapParam"), "Debug leak: {prose}");
    }

    #[test]
    /// The only shape treated as a direct store address is a bare 64-char
    /// lowercase-hex token; anything else (wrong length, uppercase, non-hex)
    /// falls through to `run_campaign`'s file-path branch instead.
    fn is_content_id_accepts_only_64_lowercase_hex() {
        assert!(is_content_id(&"a".repeat(64)));
        assert!(!is_content_id(&"A".repeat(64)));
        assert!(!is_content_id(&"a".repeat(63)));
        assert!(!is_content_id(&format!("{}g", "a".repeat(63))));
    }

    #[test]
    /// bind_axes resolves a raw axis name against the ONE wrapped param whose
    /// path ends with ".{raw}" (or equals it), producing a co-indexed point.
    fn bind_axes_resolves_a_unique_suffix_match() {
        let space = vec![spec("sma.length")];
        let params = vec![("length".to_string(), Scalar::i64(14))];
        let point = bind_axes(&space, "strat", &params).unwrap();
        assert_eq!(point, vec![Scalar::i64(14).cell()]);
    }

    #[test]
    /// A raw campaign axis that matches no open param of the wrapped
    /// strategy is a named Bind fault naming the axis, never a silently
    /// dropped binding.
    fn bind_axes_refuses_an_unmatched_axis() {
        let space = vec![spec("sma.length")];
        let params = vec![("period".to_string(), Scalar::i64(14))];
        let err = bind_axes(&space, "strat", &params).unwrap_err();
        let MemberFault::Bind(m) = err else { panic!("expected Bind, got {err:?}") };
        assert!(m.contains("period") && m.contains("matches no open param"));
    }

    #[test]
    /// A raw axis matching MORE THAN ONE wrapped param is a named Bind fault
    /// that lists every ambiguous candidate, never a silent first-match pick.
    fn bind_axes_refuses_an_ambiguous_axis() {
        let space = vec![spec("fast.length"), spec("slow.length")];
        let params = vec![("length".to_string(), Scalar::i64(14))];
        let err = bind_axes(&space, "strat", &params).unwrap_err();
        let MemberFault::Bind(m) = err else { panic!("expected Bind, got {err:?}") };
        assert!(m.contains("ambiguous") && m.contains("fast.length") && m.contains("slow.length"));
    }

    #[test]
    /// An open wrapped param left unbound by any campaign axis is a named
    /// Bind fault naming the param, not an implicit default.
    fn bind_axes_refuses_an_uncovered_param() {
        let space = vec![spec("sma.length"), spec("sma.threshold")];
        let params = vec![("length".to_string(), Scalar::i64(14))];
        let err = bind_axes(&space, "strat", &params).unwrap_err();
        let MemberFault::Bind(m) = err else { panic!("expected Bind, got {err:?}") };
        // The refusal speaks the RAW campaign-axis namespace (what the doc
        // must say), with the wrapped bind-time path as a parenthetical (#203).
        assert!(
            m.contains("open param \"threshold\"") && m.contains("(wrapped path: sma.threshold)"),
            "raw-namespace prose expected: {m}"
        );
    }

    #[test]
    /// 0109: the campaign cell key is content-derived (the `member_key`
    /// discipline — never a runtime ordinal): strategy content-id prefix +
    /// instrument + doc-positional window ordinal, one portable component.
    fn campaign_cell_key_is_content_derived() {
        let strategy = "bb34aa55".repeat(8); // a 64-hex content id
        assert_eq!(campaign_cell_key(&strategy, "GER40", 0), "bb34aa55-GER40-w0");
    }

    #[test]
    /// A non-portable instrument byte maps to `_` (the shared
    /// `sanitize_component` charset) — one path component, never a nested
    /// path or an invalid directory name.
    fn campaign_cell_key_sanitizes_non_portable_bytes() {
        let strategy = "0123abcd".repeat(8);
        assert_eq!(campaign_cell_key(&strategy, "GER/40", 3), "0123abcd-GER_40-w3");
    }

    #[test]
    /// The tap->channel routing mirrors `persist_traces_r` (equity<-eq,
    /// exposure<-ex, r_equity<-req); `net_r_equity` is unproducible on the
    /// campaign runner's cost-free wrap and routes to no channel.
    fn tap_channel_routes_the_producible_vocabulary_only() {
        assert_eq!(tap_channel("equity"), Some(TapChannel::Equity));
        assert_eq!(tap_channel("exposure"), Some(TapChannel::Exposure));
        assert_eq!(tap_channel("r_equity"), Some(TapChannel::REquity));
        assert_eq!(tap_channel("net_r_equity"), None);
    }
}