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

//! [`DefaultMemberRunner`]: the shipped `aura_campaign::MemberRunner`
//! implementation over the loaded-blueprint machinery, plus the disk-layout
//! writers that persist a campaign run's traces (#295).

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

use aura_backtest::{point_from_params, summarize, summarize_r, RunReport};
use aura_campaign::{
    CampaignOutcome, CellOutcome, CellSpec, MemberFault, MemberRunner,
};
use aura_core::{Scalar, ScalarKind, Timestamp};
use aura_engine::{blueprint_from_json, f64_field, ColumnarTrace};
use aura_ingest::{instrument_geometry, open_columns_window, unix_ms_to_epoch_ns};
use aura_registry::WriteKind;
use aura_research::{CampaignDoc, CostSpec};

use crate::axes::{bind_axes, raw_bound_overrides_of};
use crate::coverage::interior_gap_months;
use crate::member::{
    blueprint_axis_probe, blueprint_axis_probe_reopened, key_supply, reopen_all,
    run_blueprint_member, wrap_r, wrapped_bound_overrides_of, CostLeg,
};
use crate::project::Env;
use crate::translate::{cost_nodes_for, stop_rule_for_regime};

/// Map any byte outside the portable directory-name charset `[A-Za-z0-9._-]`
/// to `_`. The single source of filesystem-portability for an on-disk path
/// component (valid on Linux / Windows / macOS, also URL-path- and
/// cloud-sync-safe). Used by the campaign trace layout's per-cell/per-member
/// keys ([`campaign_cell_key`], [`member_trace_key`]) — a durable disk-layout
/// contract, not stdout presentation.
pub fn sanitize_component(s: &str) -> String {
    s.chars()
        .map(|c| if c.is_ascii_alphanumeric() || matches!(c, '.' | '_' | '-') { c } else { '_' })
        .collect()
}

/// Render a scalar value case-lessly: integers/timestamps as decimal digits,
/// bool as `true`/`false`, f64 via Rust's `Display` (decimal, shortest
/// round-trip, NO scientific notation). Case-less rendering is what keeps two
/// members of one family from ever differing only by letter case
/// (case-insensitive-FS safety). The single source for the trace layout's
/// member-key label AND the shell's `aura reproduce` rendering — the two must
/// stay byte-identical (#224).
pub fn render_value(v: &Scalar) -> String {
    match v {
        Scalar::I64(n) => n.to_string(),
        Scalar::F64(x) => x.to_string(),
        Scalar::Bool(b) => b.to_string(),
        Scalar::Timestamp(t) => t.0.to_string(),
    }
}

/// The shipped harness/data binding seam for `aura_campaign::execute`: members
/// run through the shipped loaded-blueprint machinery (`wrap_r` reduce-mode
/// via `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.
pub struct DefaultMemberRunner<'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<'a> DefaultMemberRunner<'a> {
    pub fn new(
        env: &'a Env,
        server: Arc<data_server::DataServer>,
        bindings: BTreeMap<String, String>,
        cost: Vec<CostSpec>,
    ) -> Self {
        Self { env, server, bindings, cost }
    }

    /// The runner's own data server, shared with the trace-persistence tail
    /// (`persist_campaign_traces` re-runs members over the same archive).
    pub fn server(&self) -> Arc<data_server::DataServer> {
        Arc::clone(&self.server)
    }
}

impl MemberRunner for DefaultMemberRunner<'_> {
    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 = 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 =
            blueprint_axis_probe_reopened(&cell.blueprint_json, self.env, &overrides)
                .param_space();
        let point = bind_axes(&space, &cell.strategy_id, params)?;
        let signal = 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 = 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,
        })
    }
}

/// 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 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 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.id, "equity" | "exposure" | "r_equity" | "net_r_equity")),
        "tap_vocabulary drifted from the channels persist_campaign_traces routes"
    );
    match tap {
        "equity" => Some(TapChannel::Equity),
        "exposure" => Some(TapChannel::Exposure),
        "r_equity" => Some(TapChannel::REquity),
        "net_r_equity" => Some(TapChannel::Net),
        _ => None,
    }
}

/// The note text for a requested tap outside the closed vocabulary (#278):
/// pure over just the tap name, so the exact marker-bearing text
/// `persist_campaign_traces` writes to stderr is unit-testable without
/// booting a real archive run (the `tap_channel` `None` arm that reaches
/// this is defensive — `validate_campaign` already refuses an unknown tap
/// name before this function ever runs).
fn tap_not_produced_note(tap: &str) -> String {
    format!("aura: note: tap \"{tap}\" is not produced by this run; skipped")
}

/// The note text for a cell with neither a nominee nor a non-empty terminal
/// family (#278): pure over the record's own cell fields, so the exact
/// marker-bearing text `persist_campaign_traces` writes to stderr is
/// unit-testable without booting a real archive run — the `cell_member_fanout`
/// twin above already covers the emptiness logic itself.
fn no_nominee_note(strategy: &str, instrument: &str, window_ms: (i64, i64)) -> String {
    format!(
        "aura: note: cell {strategy}/{instrument}/[{}, {}]: no nominee; no traces persisted",
        window_ms.0, window_ms.1
    )
}

/// 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}`.
pub 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}") };
    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).
pub fn member_trace_key(report: &RunReport, ordinal: usize) -> String {
    let label = report
        .manifest
        .params
        .iter()
        .map(|(n, v)| format!("{n}={}", render_value(v)))
        .collect::<Vec<_>>()
        .join(", ");
    let key = if label.is_empty() { ordinal.to_string() } else { label };
    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.
pub 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 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: note: 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!("{}", tap_not_produced_note(tap)),
        }
    }

    // `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!(
                "{}",
                no_nominee_note(&cell_rec.strategy, &cell_rec.instrument, cell_rec.window_ms)
            );
            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 = 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 = 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 =
            blueprint_axis_probe_reopened(blueprint_json, env, &overrides).param_space();
        // Same resolved pip the member ran at (`DefaultMemberRunner::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 `DefaultMemberRunner::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 = 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 = 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 `DefaultMemberRunner::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
            // `DefaultMemberRunner::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 (#295'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(|| CostLeg {
                nodes: cost_nodes_for(&campaign.cost, &cell_rec.instrument),
                tx_cost,
                tx_net,
            });
            let mut h =
                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(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_engine::RunManifest;

    /// 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]
    /// 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]
    /// #278: the unrecognized-tap note carries the `note` class marker AND
    /// names the offending tap — the exact text `persist_campaign_traces`
    /// writes to stderr on `tap_channel`'s defensive `None` arm.
    fn tap_not_produced_note_carries_the_note_marker_and_names_the_tap() {
        let msg = tap_not_produced_note("bogus");
        assert!(msg.starts_with("aura: note: "), "carries the note class marker: {msg}");
        assert!(msg.contains("\"bogus\""), "names the offending tap: {msg}");
        assert!(msg.contains("is not produced by this run"), "states the reason: {msg}");
    }

    #[test]
    /// #278: the no-candidate-cell note carries the `note` class marker AND
    /// names the cell (strategy/instrument/window) — the exact text
    /// `persist_campaign_traces` writes to stderr when `cell_member_fanout`
    /// (unit-tested above) reports no nominee and no non-empty family.
    fn no_nominee_note_carries_the_note_marker_and_names_the_cell() {
        let msg = no_nominee_note("bb34aa55", "GER40", (1_000, 2_000));
        assert!(msg.starts_with("aura: note: "), "carries the note class marker: {msg}");
        assert!(msg.contains("bb34aa55/GER40/[1000, 2000]"), "names the cell: {msg}");
        assert!(msg.contains("no nominee; no traces persisted"), "states the reason: {msg}");
    }
}