//! `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::collections::{BTreeMap, HashSet};
use std::num::NonZeroUsize;
use std::path::{Path, PathBuf};
use std::sync::{mpsc, Arc};

use aura_campaign::{CampaignOutcome, CellOutcome, CellSpec, ExecFault, MemberFault, MemberRunner};
use aura_core::{Cell, ParamSpec, PrimitiveBuilder, Scalar, ScalarKind, Timestamp};
use aura_engine::{
    blueprint_from_json, f64_field, summarize, summarize_r, ColumnarTrace, Composite,
    FamilySelection, RunReport,
};
use aura_ingest::{instrument_geometry, open_columns_window, unix_ms_to_epoch_ns};
use aura_registry::{CampaignRunRecord, WriteKind};
use aura_research::{
    campaign_to_json, content_id_of, parse_campaign, parse_process, validate_campaign,
    validate_process, CampaignDoc, CostSpec, DocRef,
};
use aura_std::{CarryCost, ConstantCost, VolSlippageCost};

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'))
}

/// The one regime->StopRule binding, shared by `CliMemberRunner::run_member`
/// and `persist_campaign_traces`'s drift-alarm re-run (#219): `None` is the
/// default vol-stop, `Some(RiskRegime::Vol { .. })` binds that regime's own
/// params. Single-sourced so the persist-side re-run structurally cannot
/// diverge from the run-side binding again (the #219 divergence class).
fn stop_rule_for_regime(regime: Option<aura_research::RiskRegime>) -> aura_composites::StopRule {
    match regime {
        None => aura_composites::StopRule::Vol {
            length: crate::R_SMA_STOP_LENGTH,
            k: crate::R_SMA_STOP_K,
        },
        Some(aura_research::RiskRegime::Vol { length, k }) => {
            aura_composites::StopRule::Vol { length, k }
        }
        Some(aura_research::RiskRegime::VolTf { period_minutes, length, k }) => {
            aura_composites::StopRule::VolTf { period_minutes, length, k }
        }
    }
}

/// The one CostSpec -> cost-node binding (#234), beside its risk sibling
/// `stop_rule_for_regime` and shared the same way: `CliMemberRunner::run_member`
/// (via `run_blueprint_member`) and `persist_campaign_traces`'s drift-alarm
/// re-run both bind through here, so the persist-side re-run structurally
/// cannot diverge from the run-side cost model (the #219 divergence class,
/// cost edition). The bound knob names are the builders' own `ParamSpec` names
/// — the `CostSpec` serde vocabulary conforms to them.
pub(crate) fn cost_nodes_for(specs: &[CostSpec], instrument: &str) -> Vec<PrimitiveBuilder> {
    specs
        .iter()
        .map(|s| {
            let (knob, v) = cost_knob(s, instrument);
            match s {
                CostSpec::Constant { .. } => ConstantCost::builder().bind(knob, Scalar::f64(v)),
                CostSpec::VolSlippage { .. } => VolSlippageCost::builder().bind(knob, Scalar::f64(v)),
                CostSpec::Carry { .. } => CarryCost::builder().bind(knob, Scalar::f64(v)),
            }
        })
        .collect()
}

/// The one CostSpec-variant -> (knob name, value) mapping, shared by
/// `cost_nodes_for`'s bind above and `run_blueprint_member`'s manifest stamp
/// (main.rs): the stamp key must equal the bind key for reproduce to
/// re-derive a `CostSpec` from a stored manifest, so both sites read the
/// name off this single function rather than each carrying its own literal.
pub(crate) fn cost_knob(spec: &CostSpec, instrument: &str) -> (&'static str, f64) {
    let (knob, value) = match spec {
        CostSpec::Constant { cost_per_trade } => ("cost_per_trade", cost_per_trade),
        CostSpec::VolSlippage { slip_vol_mult } => ("slip_vol_mult", slip_vol_mult),
        CostSpec::Carry { carry_per_cycle } => ("carry_per_cycle", carry_per_cycle),
    };
    let v = value.resolve(instrument).unwrap_or_else(|| {
        // Unreachable after intrinsic validation (map keys ≡ instruments);
        // refuse loudly rather than charging 0 silently if it ever surfaces.
        eprintln!("aura: cost {knob}: no entry for instrument {instrument}");
        std::process::exit(1);
    });
    (knob, v)
}

/// 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::SelectionFreeSweepNotTerminal { stage } => format!(
            "process stage {stage}: a sweep without a selection group (metric + \
             select) must be the last stage of its process"
        ),
        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(m) => aura_campaign::member_fault_prose(m),
        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,
}

/// Strip the wrapper's exactly-one leading node segment from a WRAPPED param
/// path, yielding the RAW campaign-axis name the doc speaks (#203): the bind
/// convention prefixes a strategy's own param paths with one root-composite
/// segment before they enter the runnable (wrapped) param space, so the raw
/// axis is everything after that segment's dot. Returns `name` unchanged if
/// there is no dot (defensive; every wrapped param space produced by
/// `blueprint_axis_probe` has one). This is the single definition of "the
/// wrapper segment" — [`raw_matches_wrapped`] is built on it so a wrap-depth
/// change cannot desync the two directions of the #203 convention.
pub(crate) fn wrapped_to_raw_axis(name: &str) -> &str {
    name.split_once('.').map(|(_, rest)| rest).unwrap_or(name)
}

/// Does a RAW campaign-axis name (the doc's own namespace) address this ONE
/// wrapped param path? True for an exact match (an unwrapped param, if one
/// ever exists) or when stripping the wrapper's one node segment
/// ([`wrapped_to_raw_axis`]) yields exactly `raw`. Inverse of
/// `wrapped_to_raw_axis`, co-located and built on it (#203): the two are the
/// only two places the wrapper-segment convention is expressed.
pub(crate) fn raw_matches_wrapped(raw: &str, wrapped: &str) -> bool {
    wrapped == raw || wrapped_to_raw_axis(wrapped) == raw
}

/// Suffix-join each raw campaign axis onto exactly one wrapped param
/// ([`raw_matches_wrapped`] — wrapped == raw, or stripping the wrapper's one
/// node segment yields raw), 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.
pub(crate) 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)| raw_matches_wrapped(raw, &p.name))
            .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 = wrapped_to_raw_axis(&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 #246 override subset of one member's campaign-axis names (already RAW
/// — a campaign document speaks the raw namespace, #203, hence the `raw_`
/// name prefix distinguishing this from main.rs's WRAPPED-input siblings):
/// every name that matches no entry of the un-reopened wrapped OPEN
/// `open_space` but names a BOUND param of `raw_signal` (`bound_param_space()`'s
/// `.name` is already path-qualified in strategy/RAW coordinates, exactly
/// like a raw campaign axis — no wrap-prefix stripping needed, unlike
/// `wrapped_bound_overrides_of`/`override_paths` in main.rs, which start from
/// WRAPPED CLI axis names). Silent like `wrapped_bound_overrides_of` (skips a
/// name matching neither space), NOT the validating `override_paths`:
/// `run_member` runs inside a sweep worker and must never call
/// `std::process::exit` — an unmatched name still surfaces as `bind_axes`'s
/// own named `MemberFault::Bind` once the (reopened) space below is built.
pub(crate) fn raw_bound_overrides_of(
    param_names: &[String],
    open_space: &[ParamSpec],
    raw_signal: &Composite,
) -> Vec<String> {
    let bound: HashSet<String> =
        raw_signal.bound_param_space().into_iter().map(|b| b.name).collect();
    param_names
        .iter()
        .filter(|raw| {
            !open_space.iter().any(|p| raw_matches_wrapped(raw, &p.name))
                && bound.contains(raw.as_str())
        })
        .cloned()
        .collect()
}

/// 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>,
    /// The campaign's `data.bindings` overrides (role -> column), threaded
    /// into per-member binding resolution (#231: rebind per campaign without
    /// touching the blueprint's content id).
    bindings: BTreeMap<String, String>,
    /// The campaign's cost model (#234), threaded into every member run via
    /// `cost_nodes_for` (empty = zero costs, today's graph).
    cost: Vec<CostSpec>,
}

impl MemberRunner for CliMemberRunner<'_> {
    fn run_member(
        &self,
        cell: &CellSpec,
        params: &[(String, Scalar)],
        window_ms: (i64, i64),
    ) -> Result<RunReport, MemberFault> {
        // The member's blueprint, loaded RAW (no re-open yet) — the SAME
        // reload the probe below wraps, so `bound_param_space()` here and
        // the probe's `param_space()` after `blueprint_axis_probe_reopened`
        // resolve against one topology.
        let signal = blueprint_from_json(&cell.blueprint_json, &|t| self.env.resolve(t))
            .expect("stored blueprint passed the referential gate; reload is infallible");

        // The un-reopened wrapped OPEN space (#246) — the SAME probe the
        // sweep verbs resolve against (single-sourced in main.rs; identical
        // `false, true, None` wrap) — derives which of this cell's RAW axis
        // names re-open a bound param before the real, reopened probe/reload
        // are built: probe and member reload must re-open identically so
        // `params` resolves against one space.
        let raw_space = crate::blueprint_axis_probe(&cell.blueprint_json, self.env).param_space();
        let param_names: Vec<String> = params.iter().map(|(n, _)| n.clone()).collect();
        let overrides = raw_bound_overrides_of(&param_names, &raw_space, &signal);
        let space =
            crate::blueprint_axis_probe_reopened(&cell.blueprint_json, self.env, &overrides)
                .param_space();
        let point = bind_axes(&space, &cell.strategy_id, params)?;
        let signal = crate::reopen_all(signal, &overrides);

        // The member's resolved input binding (campaign data.bindings
        // overrides win over name defaults). A refusal is a member fault,
        // never a process exit.
        let binding =
            crate::binding::resolve_binding(&cell.strategy_id, signal.input_roles(), &self.bindings)
                .map_err(MemberFault::Bind)?;

        // 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);
        // One source per resolved binding column, canonical order (the same
        // order `wrap_r` declares the roles in — the shared plan).
        let sources = match open_columns_window(
            &self.server,
            &cell.instrument,
            Some(from),
            Some(to),
            &binding.columns(),
        ) {
            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
        // sources whose first peek is None (open_window's documented contract)
        // — the same no-data condition (all columns decode the same bars, so
        // probing the first source covers the set).
        if aura_engine::Source::peek(sources[0].as_ref()).is_none() {
            return Err(MemberFault::NoData {
                instrument: cell.instrument.clone(),
                window_ms,
            });
        }

        // The shipped member recipe (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 stop = stop_rule_for_regime(cell.regime);
        let mut report = crate::run_blueprint_member(
            signal,
            &point,
            &space,
            sources,
            (from, to),
            0,
            geo.pip_size,
            &cell.strategy_id,
            self.env,
            stop,
            &binding,
            &self.cost,
            &cell.instrument,
        );
        report.manifest.instrument = Some(cell.instrument.clone());
        Ok(report)
    }

    /// #272: derive the cell's archive coverage from the #264 primitives —
    /// `None` when the archive has no file for the instrument at all (the
    /// `NoData` member fault's job, not coverage) or when the effective
    /// evaluated span covers the requested window exactly with no interior
    /// gap month. One call per cell (not per member): coverage is a property
    /// of the cell's (instrument, window), never a swept param point.
    fn window_coverage(&self, cell: &CellSpec) -> Option<aura_registry::CellCoverage> {
        let data_path = PathBuf::from(self.env.data_path());
        let months = aura_ingest::list_m1_months(&data_path, &cell.instrument);
        if months.is_empty() {
            return None; // no archive at all is the NoData fault's job, not coverage
        }
        let (eff_from_ts, eff_to_ts) = aura_ingest::archive_extent(
            &self.server,
            &data_path,
            &cell.instrument,
            Some(cell.window_ms.0),
            Some(cell.window_ms.1),
        )?;
        let eff_from = aura_ingest::epoch_ns_to_unix_ms(eff_from_ts);
        let eff_to = aura_ingest::epoch_ns_to_unix_ms(eff_to_ts);
        let gap_months = interior_gap_months(&months, cell.window_ms);
        if eff_from == cell.window_ms.0 && eff_to == cell.window_ms.1 && gap_months.is_empty() {
            return None; // fully covered — nothing to annotate
        }
        Some(aura_registry::CellCoverage {
            effective_from_ms: eff_from,
            effective_to_ms: eff_to,
            gap_months,
        })
    }
}

/// The whole `(year, month)`s missing INSIDE `months`' own span (mirrors the
/// interior-gap walk `data_coverage_report`, main.rs, performs over a full
/// archive listing) that also fall inside `window_ms` (Unix-ms) — one
/// `"YYYY-MM"` entry per missing month, never a collapsed range: the
/// registry's `CellCoverage::gap_months` is a flat list an aggregate counts
/// directly. `months` is assumed sorted ([`aura_ingest::list_m1_months`]'s own
/// contract).
fn interior_gap_months(months: &[(u16, u8)], window_ms: (i64, i64)) -> Vec<String> {
    let from_ym = data_server::records::unix_ms_to_year_month(window_ms.0);
    let to_ym = data_server::records::unix_ms_to_year_month(window_ms.1);
    let mut out = Vec::new();
    for pair in months.windows(2) {
        let (prev, next) = (pair[0], pair[1]);
        let mut ym = crate::next_year_month(prev);
        while ym != next {
            if ym >= from_ym && ym <= to_ym {
                out.push(crate::fmt_year_month(ym));
            }
            ym = crate::next_year_month(ym);
        }
    }
    out
}

/// Stdout shape of a campaign run. `Full` is `aura campaign run` (emit-gated
/// member/selection lines + the always-on final `campaign_run` record line).
/// `MemberLinesOnly` is dissolved-verb sugar: the generated document's
/// `emit: ["family_table"]` already limits emission to member lines; the mode
/// additionally suppresses the final record line so the verb's stdout
/// contract is reproduced byte-for-byte. The record append is identical in
/// both modes — presentation changes, the record does not.
#[derive(Clone, Copy, PartialEq)]
pub(crate) enum RunPresentation {
    Full,
    MemberLinesOnly,
}

/// `aura campaign run <target>`: resolve, gate, execute, emit. Every `Err`
/// is a refusal `campaign_cmd` renders as `aura: {msg}` + exit 1; an `Ok`
/// carries the failed-cell count (#272) so the caller can exit 3 on a
/// completed-with-failures run via `exit_on_campaign_result`.
pub(crate) fn run_campaign(
    target: &str,
    env: &Env,
    parallel_instruments: NonZeroUsize,
) -> Result<usize, 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"
        ));
    };

    run_campaign_by_id(&campaign_id, env, RunPresentation::Full, parallel_instruments)
}

/// An executed campaign plus the context its presenter and the dissolved-verb
/// sugar consume: `outcome` (records + per-cell realizations) and the
/// `campaign` / `strategies` / `server` the emit + trace-persistence tail need.
/// Returned by [`run_campaign_returning`] so a caller can read the outcome
/// without the stdout tail (a forthcoming dissolved-verb sugar path — the
/// generalize translator's own runner lands in a later task).
pub(crate) struct CampaignRun {
    pub outcome: aura_campaign::CampaignOutcome,
    pub campaign: CampaignDoc,
    pub strategies: Vec<(String, String)>,
    pub server: Arc<data_server::DataServer>,
}

/// The one campaign executor path from a resolved content id onward: 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, execute, and emit per `presentation`. Shared by
/// `run_campaign` (`RunPresentation::Full`) and the dissolved-verb sugar path
/// (`verb_sugar::run_sweep_sugar`, `RunPresentation::MemberLinesOnly`) — no
/// project gate here: the sugar path must work project-less exactly as the
/// inline verb it replaces did (store mechanics run against `env.registry()`
/// in both cases).
pub(crate) fn run_campaign_by_id(
    campaign_id: &str,
    env: &Env,
    presentation: RunPresentation,
    parallel_instruments: NonZeroUsize,
) -> Result<usize, String> {
    let run = run_campaign_returning(campaign_id, env, parallel_instruments)?;
    present_campaign(run, presentation, env)
}

/// The one campaign executor path from a resolved content id up to (not
/// including) presentation: fetch the stored canonical bytes by id (so file
/// addressing and id addressing produce the same realization by construction),
/// re-run the intrinsic tier, resolve strategies, execute — returning the
/// outcome bundled with the context the presenter needs. Shared by
/// `run_campaign_by_id` (which then presents) and a forthcoming dissolved-verb
/// sugar path that reads the outcome and self-prints (lands in a later task,
/// alongside the generalize translator's own runner).
pub(crate) fn run_campaign_returning(
    campaign_id: &str,
    env: &Env,
    parallel_instruments: NonZeroUsize,
) -> Result<CampaignRun, String> {
    let registry = env.registry();
    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())),
        bindings: campaign.data.bindings.clone(),
        cost: campaign.cost.clone(),
    };
    let outcome = aura_campaign::execute(
        campaign_id,
        &campaign,
        &process,
        &strategies,
        &runner,
        &registry,
        parallel_instruments,
    )
    .map_err(|f| exec_fault_prose(&f))?;

    Ok(CampaignRun { outcome, campaign, strategies, server: runner.server })
}

/// Emit an executed campaign's results per `presentation`: the zero-survivor
/// stderr notes, the emit-gated family/selection lines, the always-on record
/// line (Full only), then trace persistence. Split out of `run_campaign_by_id`
/// so the dissolved-verb sugar can consume the outcome without this stdout tail.
fn present_campaign(
    run: CampaignRun,
    presentation: RunPresentation,
    env: &Env,
) -> Result<usize, String> {
    let CampaignRun { outcome, campaign, strategies, server } = run;

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

    // #272: failed-cell notes (the gate-empty idiom's sibling) — a failed
    // cell never aborts the campaign; its fault is recorded and the run
    // continues over the remaining cells.
    let mut failed = 0usize;
    let mut fail_labels: Vec<String> = Vec::new();
    for cell in &outcome.record.cells {
        if let Some(f) = &cell.fault {
            failed += 1;
            fail_labels.push(format!("{}: {}", cell.instrument, cell_fault_kind_label(f.kind)));
            eprintln!(
                "aura: cell ({}, {}, [{}, {}]) failed at stage {}: {} — recorded, campaign continues",
                cell.strategy, cell.instrument, cell.window_ms.0, cell.window_ms.1, f.stage, f.detail
            );
        }
    }

    // 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")
                );
            }
        }
    }
    if presentation == RunPresentation::Full {
        println!(
            "{}",
            serde_json::to_string(&CampaignRunLine { campaign_run: &outcome.record })
                .expect("campaign run record serializes")
        );
        eprintln!(
            "campaign run {} recorded: {} cells{}",
            outcome.record.run,
            outcome.record.cells.len(),
            if failed == 0 {
                String::new()
            } else {
                format!(", {failed} failed ({})", fail_labels.join(", "))
            }
        );
    }

    // 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,
            &server,
            env,
        )?;
    }
    Ok(failed)
}

/// #272: the fault-kind label the failed-cell summary line names — the same
/// closed `CellFaultKind` vocabulary an aggregate over `campaign_runs.jsonl`
/// counts by. Debug-leak-free (a fixed lowercase-snake label, not a Debug
/// frame of the enum variant).
fn cell_fault_kind_label(kind: aura_registry::CellFaultKind) -> &'static str {
    use aura_registry::CellFaultKind as K;
    match kind {
        K::NoData => "no_data",
        K::Bind => "bind",
        K::Run => "run",
        K::Panic => "panic",
        K::Window => "window",
    }
}

/// Which drained wrap-convention channel a requested tap persists from on a
/// campaign trace re-run. The routing follows the wrap-convention channels:
/// 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 <- the cost leg's LinComb(4) net-curve recorder
/// (tx_net, #234) — produced only when the campaign document carries a cost
/// block (the caller's producibility check).
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub(crate) enum TapChannel {
    Equity,
    Exposure,
    REquity,
    Net,
}

/// Route one requested tap of the closed vocabulary to its drained channel;
/// `None` marks a name outside the vocabulary (unreachable — `validate_campaign`
/// refuses it — and mapped defensively rather than guessed at).
pub(crate) fn tap_channel(tap: &str) -> Option<TapChannel> {
    // The emit_vocabulary debug_assert's twin: every vocabulary tap must be
    // routable here — a future fifth vocabulary entry fails loudly instead of
    // silently skipping.
    debug_assert!(
        aura_research::tap_vocabulary()
            .iter()
            .all(|t| matches!(*t, "equity" | "exposure" | "r_equity" | "net_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),
        "net_r_equity" => Some(TapChannel::Net),
        _ => 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.
/// `regime_ordinal` (#219/#212) discriminates two risk regimes over the same
/// (strategy, instrument, window) cell so their traces never collide: the
/// default regime (ordinal 0) keeps the pre-#219 key unchanged (no stored
/// trace dir is renamed by this change), a non-default regime appends
/// `-r{ordinal}`.
fn campaign_cell_key(
    strategy: &str,
    instrument: &str,
    window_ordinal: usize,
    regime_ordinal: usize,
) -> String {
    let strategy8 = strategy.get(..8).unwrap_or(strategy);
    let base = format!("{strategy8}-{instrument}-w{window_ordinal}");
    let base =
        if regime_ordinal == 0 { base } else { format!("{base}-r{regime_ordinal}") };
    crate::sanitize_component(&base)
}

/// The per-member subdirectory key under a swept cell's trace dir (#224):
/// the member's own manifest params label — the exact `"name=value, ..."`
/// join `aura reproduce` prints — sanitized for filesystem use (a raw label
/// carries `=`/`, ` which `sanitize_component` maps to `_`). The member's
/// ordinal into the family is the fallback when the label is empty (a closed
/// blueprint / a monte-carlo seed member carries no tuning params).
fn member_trace_key(report: &RunReport, ordinal: usize) -> String {
    let label = report
        .manifest
        .params
        .iter()
        .map(|(n, v)| format!("{n}={}", crate::render_value(v)))
        .collect::<Vec<_>>()
        .join(", ");
    let key = if label.is_empty() { ordinal.to_string() } else { label };
    crate::sanitize_component(&key)
}

/// The per-cell member fan-out (#224): a nominee cell writes its one trace
/// directly at `<cell_key>/` (`None` subdir key, unchanged since 0109); a
/// no-nominee cell with a completed terminal family (the selection-free sweep
/// shape) writes every one of that family's members under its own
/// `<cell_key>/<member_key>/` subdirectory instead of silently dropping every
/// member but a (non-existent) nominee. An empty return (no nominee AND no
/// non-empty terminal family) is the caller's cue to skip the cell loudly.
/// Pure over the executed outcome's own `CellOutcome`, so it is unit-testable
/// without booting a real archive run.
fn cell_member_fanout(cell_out: &CellOutcome) -> Vec<(Option<String>, &RunReport)> {
    match &cell_out.nominee {
        Some((_, nominee_report)) => vec![(None, nominee_report)],
        None => match cell_out.families.last() {
            Some(fam) if !fam.reports.is_empty() => fam
                .reports
                .iter()
                .enumerate()
                .map(|(i, r)| (Some(member_trace_key(r, i)), r))
                .collect(),
            _ => Vec::new(),
        },
    }
}

/// Persist the requested `persist_taps` for every cell under
/// `traces/<trace_name>/<cell_key>/` (0109, #201 d5; #224): a cell with a
/// nominee (walkforward/generalize/mc — selection-bearing pipelines) writes
/// its single trace directly at `<cell_key>/`, unchanged since 0109. A cell
/// with NO nominee but a completed terminal family (a selection-free sweep,
/// #224's headline: "each swept member's tap series") writes EVERY member of
/// that family under its own `<cell_key>/<member_key>/` subdirectory — never
/// narrowing to one nominated member, which would silently drop the others
/// the sweep actually produced. Each written member is independently re-run
/// once in non-reduce trace mode over its own recorded `manifest.window`,
/// asserting the re-run METRICS equal the recorded member metrics (the C1
/// drift alarm — manifest fields are fresh-context and not compared), and
/// writes the requested-AND-producible taps through the sweep verbs'
/// member-dir mechanism: a slash-joined `"{name}/{key}"` handed to
/// `TraceStore::write` (the slash-joined member-dir layout the sweep verbs
/// established). The written index manifest is the RECORDED
/// member's — the trace documents that run's provenance, not a re-derived
/// fresh-context one. Loud stderr per no-candidate cell (neither a nominee
/// nor a non-empty terminal family) 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.
pub(crate) 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. `net_r_equity` is producible
    // exactly when the document carries a cost block (#234); the skip notice
    // names the remedy — the tap is optional presentation, not a refusal.
    // When nothing producible was requested, 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(TapChannel::Net) if campaign.cost.is_empty() => eprintln!(
                "aura: tap \"{tap}\" needs a cost model; add a cost block to the campaign document; skipped"
            ),
            Some(ch) => routed.push((tap.as_str(), ch)),
            None => eprintln!("aura: tap \"{tap}\" is not produced by this 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) {
        // A nominee cell writes its one trace directly at `<cell_key>/`
        // (`None` subdir key, unchanged since 0109); a no-nominee cell with a
        // completed terminal family (the selection-free sweep shape, #224)
        // writes every one of that family's members under its own
        // `<cell_key>/<member_key>/` subdirectory instead of silently
        // dropping every member but a (non-existent) nominee.
        let members = cell_member_fanout(cell_out);
        if members.is_empty() {
            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,
            cell_rec.regime_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
                )
            })?;
        // The #246 override set (silent variant, `reproduce_family_in`'s
        // recipe): re-derived from the cell's own recorded manifest param
        // names, against a raw probe space + raw strategy load — the members
        // of one cell share the same knob NAMES (only values differ), so this
        // is safe to compute once, cell-invariant, from the first member.
        let raw_space = crate::blueprint_axis_probe(blueprint_json, env).param_space();
        let raw_signal = blueprint_from_json(blueprint_json, &|t| env.resolve(t))
            .expect("stored blueprint passed the referential gate; reload is infallible");
        let recorded: Vec<String> = members
            .first()
            .map(|(_, r)| r.manifest.params.iter().map(|(n, _)| n.clone()).collect())
            .unwrap_or_default();
        let overrides = crate::wrapped_bound_overrides_of(&recorded, &raw_space, &raw_signal);
        // Cell-invariant across every member of this cell (the wrapped param
        // space depends only on the cell's blueprint; the resolved geometry
        // only on the cell's instrument) — computed once here rather than
        // redundantly inside the member loop below.
        let space =
            crate::blueprint_axis_probe_reopened(blueprint_json, env, &overrides).param_space();
        // Same resolved pip the member ran at (`CliMemberRunner::run_member`'s
        // `geo.pip_size`), else the C1 drift-alarm equality below would trip
        // spuriously on a re-run computed at a different (default) pip.
        let geo = instrument_geometry(server, &cell_rec.instrument).ok_or_else(|| {
            format!(
                "no recorded geometry for symbol '{}' at {} — refusing to re-run a \
                 real instrument with a guessed pip",
                cell_rec.instrument,
                env.data_path()
            )
        })?;

        for (member_subdir, member_report) in members.iter().cloned() {
            // Re-run the member, 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
            // member report's own `manifest.window` — already epoch-ns
            // (`run_blueprint_member` stamped the post-seam bounds), so no
            // second ms->ns crossing here.
            let point = crate::point_from_params(&space, &member_report.manifest.params);
            let (from, to) = member_report.manifest.window;
            let no_data = || {
                format!(
                    "no data for instrument {} in the member window [{}, {}] (epoch-ns)",
                    cell_rec.instrument, from.0, to.0
                )
            };
            let signal = crate::reopen_all(
                blueprint_from_json(blueprint_json, &|t| env.resolve(t))
                    .expect("stored blueprint passed the referential gate; reload is infallible"),
                &overrides,
            );
            // Campaign data.bindings overrides win over name defaults — the
            // SAME resolution the member ran under, so the C1 drift alarm
            // compares like with like.
            let binding = crate::binding::resolve_binding(
                &cell_rec.strategy,
                signal.input_roles(),
                &campaign.data.bindings,
            )?;
            let sources = open_columns_window(
                server,
                &cell_rec.instrument,
                Some(from),
                Some(to),
                &binding.columns(),
            )
            .ok_or_else(no_data)?;
            if aura_engine::Source::peek(sources[0].as_ref()).is_none() {
                return Err(no_data());
            }
            // The cell's OWN regime (#219/#212), not the hardcoded default: the
            // recorded member ran under `cell_rec.regime`, bound through the same
            // `stop_rule_for_regime` helper `CliMemberRunner::run_member` uses, so
            // the two sites cannot drift apart again (the #219 divergence class)
            // and the re-run binds the same stop the C1 drift alarm below relies on.
            let stop = stop_rule_for_regime(cell_rec.regime);
            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 campaign's OWN cost model (#234), bound the same way
            // `CliMemberRunner::run_member` binds it (via `cost_nodes_for`):
            // empty = no leg, so a cost-less campaign's re-run is unchanged.
            // A non-empty model MUST be threaded here too — the recorded
            // member ran net (Task 4's `run_blueprint_member`); leaving this
            // re-run gross would trip the C1 drift alarm below on every
            // legitimate costed campaign, not just on a real divergence.
            let (tx_cost, rx_cost) = mpsc::channel();
            let (tx_net, rx_net) = mpsc::channel();
            let cost_leg = (!campaign.cost.is_empty()).then(|| crate::CostLeg {
                nodes: cost_nodes_for(&campaign.cost, &cell_rec.instrument),
                tx_cost,
                tx_net,
            });
            let mut h =
                crate::wrap_r(signal, tx_eq, tx_ex, tx_r, tx_req, stop, false, geo.pip_size, &binding, cost_leg)
                    .bootstrap_with_cells(&point)
                    .expect("the member's point re-bootstraps (it already ran this realization)");
            // The recorded member already ran this realization once (see the
            // `.expect` immediately above); `sources` are re-opened against the
            // SAME `binding` this trace re-run resolved, so a `SourceBindError`
            // here can only be an internal wiring inconsistency, not a fresh
            // user input mistake. Panic like the sibling bootstrap invariant
            // instead of a process-exit dressed as a refusal message.
            h.run_bound(crate::key_supply(&binding, sources))
                .expect("sources re-opened against `binding` key-match that binding's own roles by construction");
            // Drain ALL SIX channels (`run_signal_r` leaves req undrained; the
            // trace path must not): eq/ex/req/net feed the taps, r+cost feed
            // 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();
            let cost_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_cost.try_iter().collect();
            let net_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_net.try_iter().collect();

            // The C1 drift alarm: metrics equality against the recorded
            // member. 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, &cost_rows));
            if rerun_metrics != member_report.metrics {
                return Err(format!(
                    "trace re-run diverged from the recorded member (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,
                        TapChannel::Net => &net_rows,
                    };
                    ColumnarTrace::from_rows(tap, &[ScalarKind::F64], rows)
                })
                .collect();
            let write_key = match &member_subdir {
                Some(sub) => format!("{trace_name}/{cell_key}/{sub}"),
                None => format!("{trace_name}/{cell_key}"),
            };
            store
                .write(&write_key, &member_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_campaign::StageFamily;
    use aura_core::ScalarKind;
    use aura_engine::RunManifest;

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

    /// A minimal `RunReport` fixture carrying exactly the manifest params
    /// `member_trace_key`/`cell_member_fanout` read; the metrics are an empty
    /// `summarize`, irrelevant to either function under test.
    fn report_with_params(params: Vec<(String, Scalar)>) -> RunReport {
        RunReport {
            manifest: RunManifest {
                commit: "test".to_string(),
                params,
                defaults: Vec::new(),
                window: (Timestamp(0), Timestamp(0)),
                seed: 0,
                broker: "test".to_string(),
                selection: None,
                instrument: None,
                topology_hash: None,
                project: None,
            },
            metrics: summarize(&[], &[]),
        }
    }

    #[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]
    /// #262: `stop_rule_for_regime` binds `RiskRegime::VolTf` to
    /// `StopRule::VolTf` field-for-field — the resolve-side half of the
    /// write/resolve pair the manifest round-trip test (main.rs) covers from
    /// the stamp side; together the two arms this regime touches
    /// (`campaign_run.rs`'s bind and `main.rs`'s stamp) are both exercised.
    fn stop_rule_for_regime_binds_vol_tf_field_for_field() {
        let regime =
            Some(aura_research::RiskRegime::VolTf { period_minutes: 60, length: 14, k: 2.0 });
        assert_eq!(
            stop_rule_for_regime(regime),
            aura_composites::StopRule::VolTf { period_minutes: 60, length: 14, k: 2.0 }
        );
    }

    #[test]
    /// #203: `wrapped_to_raw_axis` and `raw_matches_wrapped` are the inverse
    /// pair of one wrapper-segment convention — stripping the wrapper off a
    /// wrapped name must satisfy the match predicate against that SAME
    /// wrapped name, and a raw name from a DIFFERENT wrapped param must not.
    fn raw_matches_wrapped_and_wrapped_to_raw_axis_are_inverses() {
        let wrapped = "sma_signal.fast.length";
        assert_eq!(wrapped_to_raw_axis(wrapped), "fast.length");
        assert!(raw_matches_wrapped(wrapped_to_raw_axis(wrapped), wrapped));
        assert!(!raw_matches_wrapped("fast.length", "sma_signal.slow.length"));
    }

    #[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]
    /// #246: `raw_bound_overrides_of` names exactly the raw axis names that
    /// miss the un-reopened wrapped OPEN space but match a BOUND param of the
    /// raw signal (in strategy/RAW coordinates, per its own doc comment) — an
    /// axis already covered by the open space is not re-flagged as an
    /// override, and an axis matching neither space is silently skipped here
    /// (bind_axes surfaces THAT case as its own named fault once the
    /// reopened space is built downstream).
    fn raw_bound_overrides_of_selects_axes_naming_a_bound_param() {
        use aura_std::{Bias, Sma};
        let raw_signal = Composite::new(
            "fixture",
            vec![
                Sma::builder().named("fast").bind("length", Scalar::i64(2)).into(),
                Bias::builder().named("exp").into(),
            ],
            vec![],
            vec![],
            vec![],
        );
        // The un-reopened wrapped OPEN space carries one extra wrap segment
        // (the CLI's own outer wrap, mirroring `blueprint_axis_probe`'s
        // output shape) ahead of the composite's own "exp.scale" path.
        let open_space = vec![ParamSpec { name: "wrap.exp.scale".to_string(), kind: ScalarKind::F64 }];
        let param_names =
            vec!["exp.scale".to_string(), "fast.length".to_string(), "nope.thing".to_string()];
        let overrides = raw_bound_overrides_of(&param_names, &open_space, &raw_signal);
        assert_eq!(
            overrides,
            vec!["fast.length".to_string()],
            "only the bound-param axis is selected; the already-open axis and the \
             unmatched axis are excluded"
        );
    }

    #[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, 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, 0), "0123abcd-GER_40-w3");
    }

    #[test]
    /// A non-default regime_ordinal (#219/#212) appends `-r{ordinal}` so two
    /// regimes over the same (strategy, instrument, window) cell land in
    /// distinct trace dirs; ordinal 0 (the default regime) is covered above
    /// and stays unsuffixed for pre-#219 key stability.
    fn campaign_cell_key_appends_regime_suffix_for_non_default_ordinal() {
        let strategy = "bb34aa55".repeat(8);
        assert_eq!(campaign_cell_key(&strategy, "GER40", 0, 2), "bb34aa55-GER40-w0-r2");
    }

    #[test]
    /// #224: `member_trace_key` renders the SAME `"name=value, ..."` join
    /// `aura reproduce` prints for a member's params (single-sourced label
    /// format), then sanitizes it for filesystem use — a raw label's `=` and
    /// `, ` separators are hostile to a path component and must map to `_`
    /// (the shared `sanitize_component` charset), never survive verbatim.
    fn member_trace_key_mirrors_reproduce_label_and_sanitizes_it() {
        let report = report_with_params(vec![
            ("fast".to_string(), Scalar::i64(2)),
            ("slow".to_string(), Scalar::i64(4)),
        ]);
        // The raw reproduce-format label is "fast=2, slow=4"; sanitizing maps
        // each of `=`, `,`, ` ` individually to `_` (so the ", " separator
        // becomes two underscores, not one).
        assert_eq!(member_trace_key(&report, 0), "fast_2__slow_4");
    }

    #[test]
    /// #224: a member with no tuning params (a closed blueprint, or a
    /// monte-carlo seed member) renders an empty reproduce label, so
    /// `member_trace_key` falls back to the member's own ordinal into the
    /// family — never an empty (invalid) path component.
    fn member_trace_key_falls_back_to_the_ordinal_when_params_are_empty() {
        let report = report_with_params(Vec::new());
        assert_eq!(member_trace_key(&report, 3), "3");
    }

    #[test]
    /// #224: two members with different params must land at distinct keys —
    /// the whole point of per-member subdirectories is that the sweep's
    /// members never collide into one on-disk slot.
    fn member_trace_key_differs_across_members_with_different_params() {
        let a = report_with_params(vec![("length".to_string(), Scalar::i64(10))]);
        let b = report_with_params(vec![("length".to_string(), Scalar::i64(20))]);
        assert_ne!(member_trace_key(&a, 0), member_trace_key(&b, 1));
    }

    /// A no-nominee `CellOutcome` whose terminal family holds two members
    /// (a plain selection-free sweep, #224's headline shape).
    fn two_member_family_cell() -> CellOutcome {
        CellOutcome {
            families: vec![StageFamily {
                stage: 0,
                block: "std::sweep",
                family_id: "fam".to_string(),
                reports: vec![
                    report_with_params(vec![("length".to_string(), Scalar::i64(10))]),
                    report_with_params(vec![("length".to_string(), Scalar::i64(20))]),
                ],
            }],
            selections: Vec::new(),
            nominee: None,
        }
    }

    #[test]
    /// #224: a no-nominee cell with a non-empty terminal family fans out to
    /// ONE (subdir, report) pair PER member — never narrowing to a single
    /// nominated member, which would silently drop every other member the
    /// sweep actually produced. Each pair carries a distinct `Some(key)`
    /// subdir (never `None`, which is reserved for the nominee case).
    fn cell_member_fanout_yields_one_entry_per_family_member() {
        let cell = two_member_family_cell();
        let members = cell_member_fanout(&cell);
        assert_eq!(members.len(), 2, "one dir per member, not one nominee");
        let Some(key0) = &members[0].0 else { panic!("member 0 must carry a subdir key") };
        let Some(key1) = &members[1].0 else { panic!("member 1 must carry a subdir key") };
        assert_ne!(key0, key1, "distinct members must land at distinct subdirs");
    }

    #[test]
    /// #224: a nominee cell (walkforward/generalize/mc) still writes its one
    /// trace directly at `<cell_key>/` — a `None` subdir key, unchanged since
    /// 0109 — regardless of how many stage families it also carries.
    fn cell_member_fanout_prefers_the_nominee_with_no_subdir() {
        let mut cell = two_member_family_cell();
        let nominee_report = report_with_params(vec![("length".to_string(), Scalar::i64(30))]);
        cell.nominee = Some((vec![("length".to_string(), Scalar::i64(30))], nominee_report));
        let members = cell_member_fanout(&cell);
        assert_eq!(members.len(), 1, "the nominee is the cell's single trace");
        assert!(members[0].0.is_none(), "the nominee writes directly at <cell_key>/, no subdir");
    }

    #[test]
    /// #224: a cell with neither a nominee nor a non-empty terminal family
    /// (every family is empty, or there are none) fans out to nothing — the
    /// caller's cue to skip the cell loudly rather than write an empty dir.
    fn cell_member_fanout_is_empty_with_no_nominee_and_no_members() {
        let cell = CellOutcome { families: Vec::new(), selections: Vec::new(), nominee: None };
        assert!(cell_member_fanout(&cell).is_empty());
    }

    #[test]
    /// #234 — a DELIBERATE pin move (was: `net_r_equity` routes to `None` on
    /// the cost-free wrap): the full closed tap vocabulary now routes,
    /// `net_r_equity` to the cost leg's net-curve channel. Producibility
    /// (does THIS run carry a cost model?) is the caller's check, not the
    /// routing's.
    fn tap_channel_routes_the_full_vocabulary() {
        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"), Some(TapChannel::Net));
        assert_eq!(tap_channel("bogus"), None);
    }

    #[test]
    /// #234: the one CostSpec -> builder binding maps each component to its
    /// shipped cost node with the knob BOUND — a bound component adds no open
    /// param, so the wrapped param_space stays cost-invariant (the probe/member
    /// space equivalence the campaign path relies on).
    fn cost_nodes_for_maps_each_component_to_its_bound_builder() {
        let nodes = cost_nodes_for(
            &[
                aura_research::CostSpec::Constant {
                    cost_per_trade: aura_research::CostValue::Scalar(2.0),
                },
                aura_research::CostSpec::VolSlippage {
                    slip_vol_mult: aura_research::CostValue::Scalar(0.5),
                },
                aura_research::CostSpec::Carry {
                    carry_per_cycle: aura_research::CostValue::Scalar(0.1),
                },
            ],
            "GER40",
        );
        let labels: Vec<String> = nodes.iter().map(|n| n.label()).collect();
        assert_eq!(labels, ["ConstantCost", "VolSlippageCost", "CarryCost"]);
        assert!(
            nodes.iter().all(|n| n.params().is_empty()),
            "every component must be fully bound (no open param leaks into param_space)"
        );
        assert!(cost_nodes_for(&[], "GER40").is_empty(), "an empty model binds no nodes");
    }

    #[test]
    /// #234: the manifest-stamp/reproduce round-trip (`cost[k].<knob>` ->
    /// `cost_specs_from_params`) reconstructs a component from its knob NAME —
    /// sound only while every shipped cost node declares exactly one,
    /// distinctly-named knob. A second knob (or a name collision) would break
    /// variant reconstruction with no compiler error; this pin makes it loud.
    fn every_cost_builder_declares_exactly_one_distinct_knob() {
        let knobs: Vec<String> = [
            aura_std::ConstantCost::builder(),
            aura_std::VolSlippageCost::builder(),
            aura_std::CarryCost::builder(),
        ]
        .iter()
        .map(|b| {
            let params = b.params();
            assert_eq!(params.len(), 1, "one knob per cost node: {}", b.label());
            params[0].name.clone()
        })
        .collect();
        let mut dedup = knobs.clone();
        dedup.sort();
        dedup.dedup();
        assert_eq!(dedup.len(), knobs.len(), "knob names must be pairwise distinct: {knobs:?}");
    }

    #[test]
    /// #272: `interior_gap_months` names a whole calendar month missing
    /// BETWEEN two present archive months (the Copper failure shape) when the
    /// requested window spans across it — not just when the window's own
    /// bounds fall short of full coverage. GAPSYM-shaped input (files at
    /// 2024-01..02 and 2024-05..06, absent 03..04) over a Jan-through-June
    /// window must name exactly the two missing interior months.
    /// #272: the summary label is a CHECKED mirror of the persisted serde form
    /// — an aggregate over `campaign_runs.jsonl` counts by the serialized
    /// `CellFaultKind` string, so the human-facing label must be exactly that
    /// string (no second source of truth that could silently drift).
    fn cell_fault_kind_label_matches_the_serialized_kind() {
        use aura_registry::CellFaultKind as K;
        for kind in [K::NoData, K::Bind, K::Run, K::Panic, K::Window] {
            let serde_form = serde_json::to_string(&kind).expect("serialize kind");
            let serde_form = serde_form.trim_matches('"');
            assert_eq!(
                cell_fault_kind_label(kind),
                serde_form,
                "the summary label must equal the persisted serde string"
            );
        }
    }

    #[test]
    fn interior_gap_months_names_a_whole_missing_month_spanned_by_the_window() {
        let months = [(2024u16, 1u8), (2024, 2), (2024, 5), (2024, 6)];
        // 2024-01-01T00:00:00Z .. 2024-07-01T00:00:00Z (Unix ms) — spans every
        // present month plus the March/April hole.
        let window_ms = (1_704_067_200_000i64, 1_719_792_000_000i64);
        let gaps = interior_gap_months(&months, window_ms);
        assert_eq!(
            gaps,
            vec!["2024-03".to_string(), "2024-04".to_string()],
            "both interior gap months must be named, in order"
        );
    }
}
