Aura/crates/aura-campaign/src/exec.rs

//! Campaign execution (#198): the cell loop over strategy x instrument x
//! window, the v1 stage semantics (sweep members -> family -> selection;
//! per-member gate filtering; the walk-forward seam), and the realization
//! record. Precondition: the CALLER has run the doc tiers (intrinsic +
//! referential validation), so this module re-checks nothing a validated
//! document guarantees (e.g. non-empty axes); execution-level refusals live
//! in the crate's preflight, which `execute` runs before any member runs.

use std::collections::BTreeMap;
use std::sync::Mutex;

use aura_analysis::{FamilySelection, SelectionMode};
use aura_core::{zip_params, Cell, ParamSpec, Scalar, Timestamp};
use aura_engine::{
    r_bootstrap, sweep, walk_forward, GridSpace, ListSpace, RollMode, RunManifest, RunMetrics,
    RunReport, Space, SweepFamily, SweepPoint, WalkForwardError, WindowBounds, WindowRoller,
    WindowRun,
};
use aura_registry::{
    derive_trace_name, generalization, optimize, optimize_deflated, optimize_plateau,
    sweep_member_reports, CampaignGeneralization, CampaignRunRecord, CellRealization, FamilyKind,
    PlateauMode, Registry, StageBootstrap, StageRealization, StageSelection,
};
use aura_research::{Axis, CampaignDoc, DocRef, ProcessDoc, SelectRule, StageBlock, WfMode};

use crate::{
    member_metric, predicate_holds, preflight, CellSpec, ExecFault, MemberFault, MemberRunner,
    DEFLATION_BLOCK_LEN, DEFLATION_N_RESAMPLES,
};

/// One family-producing stage's full member payload — what an emit renderer
/// (`family_table`) prints, in ordinal order.
#[derive(Clone, Debug)]
pub struct StageFamily {
    pub stage: usize,
    pub block: &'static str,
    pub family_id: String,
    pub reports: Vec<RunReport>,
}

/// One selection-bearing stage's payload — what an emit renderer
/// (`selection_report`) prints.
#[derive(Clone, Debug)]
pub struct StageSelectionOut {
    pub stage: usize,
    pub block: &'static str,
    pub family_id: String,
    pub winner_ordinal: usize,
    pub params: Vec<(String, Scalar)>,
    pub selection: FamilySelection,
}

/// Per-cell stage payloads, for the consumer's emit rendering.
#[derive(Clone, Debug)]
pub struct CellOutcome {
    pub families: Vec<StageFamily>,
    pub selections: Vec<StageSelectionOut>,
    /// The cell's nominated generalize candidate (spec-fixed): the LAST
    /// walk-forward window's OOS report + chosen params when a wf stage ran
    /// (the freshest out-of-sample evidence for the finally-selected params),
    /// else the sweep stage's winner; `None` for gate-truncated cells.
    pub nominee: Option<(Vec<(String, Scalar)>, RunReport)>,
}

/// Everything one campaign run produced: the stored realization record, the
/// assigned run counter, and the per-cell payloads.
#[derive(Clone, Debug)]
pub struct CampaignOutcome {
    pub record: CampaignRunRecord,
    pub run: usize,
    pub cells: Vec<CellOutcome>,
}

/// Execute a validated campaign document's process pipeline once per
/// (strategy, instrument, window) cell, in doc order, sequentially (C1:
/// parallelism lives inside the engine `sweep`, across members, never across
/// cells). `campaign_id` is the campaign document's 64-hex content id (its
/// first 8 chars prefix family names); `strategies` is index-aligned with
/// `campaign.strategies` as (blueprint content id, canonical blueprint json).
/// Writes one family per family-producing stage plus one campaign-run record;
/// a `MemberFault` aborts the whole run before any write of the faulted stage.
pub fn execute(
    campaign_id: &str,
    campaign: &CampaignDoc,
    process: &ProcessDoc,
    strategies: &[(String, String)],
    runner: &dyn MemberRunner,
    registry: &Registry,
) -> Result<CampaignOutcome, ExecFault> {
    preflight(process, campaign)?;
    if strategies.len() != campaign.strategies.len() {
        return Err(ExecFault::PipelineShape {
            detail: format!(
                "resolved strategies ({}) are not index-aligned with campaign.strategies ({})",
                strategies.len(),
                campaign.strategies.len(),
            ),
        });
    }
    let process_id = match &campaign.process.r#ref {
        DocRef::ContentId(id) | DocRef::IdentityId(id) => id.clone(),
    };
    // The store's self-keying always yields a 64-hex content id; anything else
    // is a caller bug — refuse it typed instead of panicking on the prefix
    // slice below.
    if campaign_id.len() != 64
        || !campaign_id.bytes().all(|b| matches!(b, b'0'..=b'9' | b'a'..=b'f'))
    {
        return Err(ExecFault::PipelineShape {
            detail: format!(
                "campaign_id must be a 64-hex content id (got {} chars)",
                campaign_id.len()
            ),
        });
    }
    let campaign_prefix = &campaign_id[..8];

    let mut cells_out: Vec<CellOutcome> = Vec::new();
    let mut cells_rec: Vec<CellRealization> = Vec::new();
    // Per-(strategy_ordinal, window_ordinal) nominees across instruments, in
    // cell-loop (instrument) order — the campaign-scope generalize input.
    type Nominee = Option<(Vec<(String, Scalar)>, RunReport)>;
    let mut nominees: BTreeMap<(usize, usize, usize), Vec<(String, Nominee)>> = BTreeMap::new();
    // Absent/empty risk = a single default cell (regime None, ordinal 0); a
    // non-empty list maps each regime to Some. The default's value is the member
    // runner's — the doc is never mutated (its content id hashes with risk absent).
    let regimes: Vec<(usize, Option<aura_research::RiskRegime>)> = if campaign.risk.is_empty() {
        vec![(0, None)]
    } else {
        campaign.risk.iter().copied().enumerate().map(|(i, r)| (i, Some(r))).collect()
    };
    for (strategy_ordinal, (entry, (strategy_id, blueprint_json))) in
        campaign.strategies.iter().zip(strategies).enumerate()
    {
        for instrument in &campaign.data.instruments {
            for (window_ordinal, window) in campaign.data.windows.iter().enumerate() {
                for (regime_ordinal, regime) in &regimes {
                    let cell = CellSpec {
                        strategy_ordinal,
                        strategy_id: strategy_id.clone(),
                        blueprint_json: blueprint_json.clone(),
                        axes: entry.axes.clone(),
                        instrument: instrument.clone(),
                        window_ms: (window.from_ms, window.to_ms),
                        regime: *regime,
                        regime_ordinal: *regime_ordinal,
                    };
                    let (outcome, realization) = run_cell(
                        &cell,
                        process,
                        campaign_prefix,
                        window_ordinal,
                        campaign.seed,
                        runner,
                        registry,
                    )?;
                    nominees
                        .entry((strategy_ordinal, window_ordinal, *regime_ordinal))
                        .or_default()
                        .push((instrument.clone(), outcome.nominee.clone()));
                    cells_out.push(outcome);
                    cells_rec.push(realization);
                }
            }
        }
    }

    // Campaign-scope generalize (#200 d2): for each (strategy, window), grade
    // the per-instrument nominees via the shipped `generalization`; fewer than
    // 2 winners is a recorded shortfall, never computed around.
    let generalize_metric = process.pipeline.iter().find_map(|s| match s {
        StageBlock::Generalize { metric } => Some(metric.clone()),
        _ => None,
    });
    let mut generalizations: Vec<CampaignGeneralization> = Vec::new();
    if let Some(metric) = generalize_metric {
        for ((strategy_ordinal, window_ordinal, regime_ordinal), cells) in &nominees {
            let winners: Vec<(String, Vec<(String, Scalar)>)> = cells
                .iter()
                .filter_map(|(inst, nom)| {
                    nom.as_ref().map(|(params, _)| (inst.clone(), params.clone()))
                })
                .collect();
            let missing: Vec<String> = cells
                .iter()
                .filter(|(_, nom)| nom.is_none())
                .map(|(inst, _)| inst.clone())
                .collect();
            let pairs: Vec<(String, &RunReport)> = cells
                .iter()
                .filter_map(|(inst, nom)| nom.as_ref().map(|(_, report)| (inst.clone(), report)))
                .collect();
            let grade = if pairs.len() >= 2 {
                Some(generalization(&pairs, &metric).map_err(ExecFault::Registry)?)
            } else {
                None
            };
            generalizations.push(CampaignGeneralization {
                strategy_ordinal: *strategy_ordinal,
                window_ordinal: *window_ordinal,
                regime_ordinal: *regime_ordinal,
                generalization: grade,
                winners,
                missing,
            });
        }
    }

    let mut record = CampaignRunRecord {
        campaign: campaign_id.to_string(),
        process: process_id,
        run: 0,
        seed: campaign.seed,
        cells: cells_rec,
        generalizations,
        // The claim sentinel (#201 d5): a `Some` (any content) tells
        // `append_campaign_run` to compose the derived trace name onto the
        // stored line; `None` means the document requests no taps.
        trace_name: (!campaign.presentation.persist_taps.is_empty()).then(String::new),
    };
    let run = registry.append_campaign_run(&record).map_err(ExecFault::Registry)?;
    // Mirror the store-side derivation onto the returned copy (the 0109
    // contract: stored line and returned record carry the same
    // `Some("{campaign8}-{run}")` iff persist_taps is non-empty, else None) —
    // keyed off the claim already stamped above, via the registry's own
    // `derive_trace_name` (the one source of truth for the composition), not
    // a re-evaluation of `persist_taps`.
    let trace_name = record.trace_name.is_some().then(|| derive_trace_name(campaign_id, run));
    record.run = run;
    record.trace_name = trace_name;
    Ok(CampaignOutcome { record, run, cells: cells_out })
}

/// The inter-stage population crossing `run_cell`'s seam: parameter points
/// only (from `std::grid`) or executed members (from `std::sweep`). Preflight
/// guarantees report-consuming stages (`std::gate`, `std::monte_carlo`'s
/// per-survivor arm) never see `Points`; the defensive faults below keep that
/// invariant loud if the shape rules ever drift.
enum StageFlow {
    /// (ordinal into the nearest preceding family_id-bearing stage's family,
    /// param point, member report) — the shape preflight established before
    /// std::sweep populated it.
    Members(Vec<(usize, Vec<Scalar>, RunReport)>),
    Points(Vec<Vec<Scalar>>),
}

impl StageFlow {
    /// Every flow can cross the seam as bare points — the walk_forward need.
    fn points(&self) -> Vec<Vec<Scalar>> {
        match self {
            StageFlow::Points(points) => points.clone(),
            StageFlow::Members(members) => {
                members.iter().map(|(_, point, _)| point.clone()).collect()
            }
        }
    }
}

/// Run one cell's pipeline: stages in doc order, a realized prefix that stops
/// at an empty gate (decision 8 — the cell truncates, the campaign continues).
fn run_cell(
    cell: &CellSpec,
    process: &ProcessDoc,
    campaign_prefix: &str,
    window_ordinal: usize,
    seed: u64,
    runner: &dyn MemberRunner,
    registry: &Registry,
) -> Result<(CellOutcome, CellRealization), ExecFault> {
    let grid = enumerate_grid(&cell.axes);
    let mut families: Vec<StageFamily> = Vec::new();
    let mut selections: Vec<StageSelectionOut> = Vec::new();
    let mut stages: Vec<StageRealization> = Vec::new();
    // The surviving population crossing the stage seam (see `StageFlow`).
    let mut flow = StageFlow::Members(Vec::new());
    // The cell's nominated generalize candidate: last wf window's OOS report
    // when a walk_forward ran, else the sweep winner; None when truncated.
    let mut nominee: Option<(Vec<(String, Scalar)>, RunReport)> = None;

    for (stage_ordinal, stage) in process.pipeline.iter().enumerate() {
        match stage {
            StageBlock::Grid => {
                // Enumerate-only (#256 fork B): the grid's parameter points
                // cross the seam; nothing executes, nothing persists.
                flow = StageFlow::Points(grid.points.clone());
                stages.push(StageRealization {
                    block: "std::grid".to_string(),
                    family_id: None,
                    survivor_ordinals: None,
                    selection: None,
                    bootstrap: None,
                    window_faults: vec![],
                });
            }
            StageBlock::Sweep { selection } => {
                // Deterministic, self-describing family name (the "-{run}"
                // suffix is appended by the registry).
                let family_name = format!(
                    "{campaign_prefix}-{}-{}-w{window_ordinal}-r{}-s{stage_ordinal}",
                    cell.strategy_ordinal, cell.instrument, cell.regime_ordinal,
                );
                let family = match run_members(cell, &grid, runner) {
                    Ok(family) => family,
                    Err(fault) => {
                        // Registry etc. still aborts (contain's Err arm).
                        let cf = contain(fault, stage_ordinal)?;
                        return Ok(failed_cell(cell, stages, families, selections, cf, runner));
                    }
                };
                let family_id = registry
                    .append_family(&family_name, FamilyKind::Sweep, &sweep_member_reports(&family))
                    .map_err(ExecFault::Registry)?;
                match selection {
                    Some(sel) => {
                        let (winner, selection) = select_sweep_winner(
                            &family,
                            &grid.axis_lens,
                            &sel.metric,
                            sel.select,
                            sel.deflate,
                            seed,
                        )?;
                        let winner_ordinal = family
                            .points
                            .iter()
                            .position(|p| p == &winner)
                            .expect("the winner is a member of its own family");
                        let params = zip_params(&family.space, &winner.params);
                        selections.push(StageSelectionOut {
                            stage: stage_ordinal,
                            block: "std::sweep",
                            family_id: family_id.clone(),
                            winner_ordinal,
                            params: params.clone(),
                            selection: selection.clone(),
                        });
                        // Nominate the sweep winner (superseded by a later
                        // walk_forward stage; cleared by an empty gate).
                        nominee = Some((params.clone(), winner.report.clone()));
                        stages.push(StageRealization {
                            block: "std::sweep".to_string(),
                            family_id: Some(family_id.clone()),
                            survivor_ordinals: None,
                            selection: Some(StageSelection { winner_ordinal, params, selection }),
                            bootstrap: None,
                            window_faults: vec![],
                        });
                    }
                    None => {
                        // Selection-free sweep (terminal, preflight-checked):
                        // the family is the whole result — no StageSelection
                        // is recorded (recording one would fabricate intent),
                        // no nominee is produced.
                        stages.push(StageRealization {
                            block: "std::sweep".to_string(),
                            family_id: Some(family_id.clone()),
                            survivor_ordinals: None,
                            selection: None,
                            bootstrap: None,
                            window_faults: vec![],
                        });
                    }
                }
                // The whole family flows on (decision 3: `select` names the
                // recorded selection, never a filter).
                flow = StageFlow::Members(
                    family
                        .points
                        .iter()
                        .enumerate()
                        .map(|(i, p)| (i, scalar_point(&family.space, &p.params), p.report.clone()))
                        .collect(),
                );
                families.push(StageFamily {
                    stage: stage_ordinal,
                    block: "std::sweep",
                    family_id,
                    reports: family.points.into_iter().map(|p| p.report).collect(),
                });
            }
            StageBlock::Gate { all } => {
                // A member survives iff ALL predicates hold; the crate-root
                // `predicate_holds` compares the resolved per-member metric.
                // An unresolvable metric (an R name against `metrics.r ==
                // None`) fails the member — conservative and deterministic
                // (the metric NAME was already preflighted against
                // `PER_MEMBER_METRICS`).
                let StageFlow::Members(members) = &mut flow else {
                    return Err(ExecFault::PipelineShape {
                        detail: format!(
                            "stage {stage_ordinal}: std::gate needs executed \
                             members (preflight admits no std::grid predecessor)"
                        ),
                    });
                };
                members.retain(|(_, _, report)| {
                    all.iter().all(|p| {
                        member_metric(report, &p.metric)
                            .is_some_and(|value| predicate_holds(&p.cmp, value, p.value))
                    })
                });
                let ordinals: Vec<usize> = members.iter().map(|(i, _, _)| *i).collect();
                let empty = ordinals.is_empty();
                stages.push(StageRealization {
                    block: "std::gate".to_string(),
                    family_id: None,
                    survivor_ordinals: Some(ordinals),
                    selection: None,
                    bootstrap: None,
                    window_faults: vec![],
                });
                if empty {
                    nominee = None; // a truncated cell nominates no candidate
                    break; // decision 8: truncate this cell, keep running cells
                }
            }
            StageBlock::WalkForward { .. } => {
                // Deterministic, self-describing family name (the "-{run}"
                // suffix is appended by the registry).
                let family_name = format!(
                    "{campaign_prefix}-{}-{}-w{window_ordinal}-r{}-s{stage_ordinal}",
                    cell.strategy_ordinal, cell.instrument, cell.regime_ordinal,
                );
                // The seam crossing is plain points: the walk-forward stage
                // re-sweeps the surviving param points and never needs the
                // gate bookkeeping's ordinals or reports.
                let survivor_points: Vec<Vec<Scalar>> = flow.points();
                match run_walk_forward_stage(
                    seed,
                    cell,
                    stage_ordinal,
                    stage,
                    &grid.specs,
                    &grid.axis_lens,
                    &survivor_points,
                    &family_name,
                    runner,
                    registry,
                ) {
                    Ok(WfStageOutcome::Ran { family: fam, window_faults }) => {
                        stages.push(StageRealization {
                            block: "std::walk_forward".to_string(),
                            family_id: Some(fam.family_id.clone()),
                            survivor_ordinals: None,
                            selection: None,
                            bootstrap: None,
                            window_faults,
                        });
                        // Nominate the LAST window's OOS report (roll order,
                        // over the surviving folds) with its chosen params —
                        // the wf stage stamps them on the fresh OOS report's
                        // manifest.
                        nominee = fam
                            .reports
                            .last()
                            .map(|last| (last.manifest.params.clone(), last.clone()));
                        families.push(fam);
                    }
                    Ok(WfStageOutcome::CellFailed(cf)) => {
                        return Ok(failed_cell(cell, stages, families, selections, cf, runner));
                    }
                    Err(e) => return Err(e), // Registry / global refusal
                }
            }
            StageBlock::MonteCarlo { resamples, block_len } => {
                // Terminal annotator (#200 d1/d3): nothing flows out. Dual
                // input: after a walk_forward, ONE bootstrap over the pooled
                // per-window OOS trade-R series (roll order); after
                // sweep/gates, one bootstrap per surviving member. Seeded from
                // the campaign seed (the deflation convention, C1).
                let bootstrap =
                    match families.iter().rev().find(|f| f.block == "std::walk_forward") {
                        Some(fam) => StageBootstrap::PooledOos(r_bootstrap(
                            &pooled_net_trade_rs(&fam.reports),
                            *resamples as usize,
                            *block_len as usize,
                            seed,
                        )),
                        // Zero-trade members (r: None) feed an empty slice —
                        // the engine's defined all-zero degenerate, recorded
                        // explicitly (a null result is a valid result).
                        None => {
                            let StageFlow::Members(members) = &flow else {
                                return Err(ExecFault::PipelineShape {
                                    detail: format!(
                                        "stage {stage_ordinal}: std::monte_carlo \
                                         per-survivor bootstrap needs executed members \
                                         (preflight admits no std::grid predecessor)"
                                    ),
                                });
                            };
                            StageBootstrap::PerSurvivor(
                                members
                                    .iter()
                                    .map(|(ordinal, _, report)| {
                                        let rs = report
                                            .metrics
                                            .r
                                            .as_ref()
                                            .map(|r| r.net_trade_rs.as_slice())
                                            .unwrap_or(&[]);
                                        (
                                            *ordinal,
                                            r_bootstrap(
                                                rs,
                                                *resamples as usize,
                                                *block_len as usize,
                                                seed,
                                            ),
                                        )
                                    })
                                    .collect(),
                            )
                        }
                    };
                stages.push(StageRealization {
                    block: "std::monte_carlo".to_string(),
                    family_id: None,
                    survivor_ordinals: None,
                    selection: None,
                    bootstrap: Some(bootstrap),
                    window_faults: vec![],
                });
            }
            StageBlock::Generalize { .. } => {
                // Campaign-scope stage (#200 d2): `execute` runs it AFTER the
                // cell loop, over the per-cell nominees across instruments —
                // nothing realizes per cell.
            }
        }
    }

    Ok((
        CellOutcome { families, selections, nominee },
        CellRealization {
            strategy: cell.strategy_id.clone(),
            instrument: cell.instrument.clone(),
            window_ms: cell.window_ms,
            regime: cell.regime,
            regime_ordinal: cell.regime_ordinal,
            fault: None,
            coverage: runner.window_coverage(cell),
            stages,
        },
    ))
}

/// The enumerated sweep grid of one cell: param specs (axis name = key, kind =
/// axis kind) in BTreeMap (lexicographic) key order, the per-axis radixes, and
/// the odometer point list — LAST axis fastest, the `GridSpace` order
/// discipline (`optimize_plateau` reads `axis_lens` in exactly this order).
struct SweepGrid {
    specs: Vec<ParamSpec>,
    axis_lens: Vec<usize>,
    points: Vec<Vec<Scalar>>,
}

/// Odometer enumeration over the axes map, delegated to the engine's own
/// `GridSpace` (last axis fastest) rather than a hand-synced second copy of
/// the odometer — single-sourcing the order `optimize_plateau` reads
/// `axis_lens` in. Zero axes yield the single empty point. Empty VALUE lists
/// cannot reach here (`validate_campaign` refuses `EmptyAxis` upstream),
/// hence the `expect`.
fn enumerate_grid(axes: &BTreeMap<String, Axis>) -> SweepGrid {
    let specs: Vec<ParamSpec> = axes
        .iter()
        .map(|(name, axis)| ParamSpec { name: name.clone(), kind: axis.kind })
        .collect();
    let values: Vec<Vec<Scalar>> = axes.values().map(|axis| axis.values.clone()).collect();
    let grid = GridSpace::new(&specs, values)
        .expect("axes validated upstream (validate_campaign refuses EmptyAxis)");
    let axis_lens = grid.axis_lens();
    let points: Vec<Vec<Scalar>> = grid
        .points()
        .into_iter()
        .map(|cells| cells.iter().zip(&specs).map(|(c, ps)| Scalar::from_cell(ps.kind, *c)).collect())
        .collect();
    SweepGrid { specs, axis_lens, points }
}

/// Ref-counted panic-hook silencer: while at least one `SilencedPanic` guard
/// is alive, the process's panic hook is a no-op, so a member panic CONTAINED
/// by one of this module's three `catch_unwind` sites (recorded as a
/// `MemberFault`/`WindowFault`, the run continuing past it) does not also
/// paint a crash-looking default backtrace to stderr — "recorded, campaign
/// continues" stays observably true, not just true in the record. Ref-counted
/// rather than a bare set/restore because `sweep`/`walk_forward` run these
/// closures on multiple OS threads concurrently (C1 disjoint-parallel
/// members): a naive per-closure set_hook/take_hook pair would race two
/// threads' install/restore against each other. The previous hook (whatever
/// it was — default or a caller's own) is saved on the 0→1 transition and
/// restored on the 1→0 transition, so this never permanently clobbers a
/// hook installed above this module.
struct SilencedPanic;

#[allow(clippy::type_complexity)]
static PANIC_HOOK_STATE: Mutex<(usize, Option<Box<dyn Fn(&std::panic::PanicHookInfo<'_>) + Sync + Send>>)> =
    Mutex::new((0, None));

impl SilencedPanic {
    /// Enter the silenced section; drop the returned guard to leave it. Hold
    /// this only around the `catch_unwind` call itself, never the whole
    /// member-run closure, so a genuinely uncontained escape (this module's
    /// own bug, not the member's) still surfaces normally once the guard has
    /// dropped and unwinding continues past `catch_unwind`.
    fn enter() -> Self {
        let mut state = PANIC_HOOK_STATE.lock().expect("panic hook state lock");
        if state.0 == 0 {
            state.1 = Some(std::panic::take_hook());
            std::panic::set_hook(Box::new(|_| {}));
        }
        state.0 += 1;
        SilencedPanic
    }
}

impl Drop for SilencedPanic {
    fn drop(&mut self) {
        let mut state = PANIC_HOOK_STATE.lock().expect("panic hook state lock");
        state.0 -= 1;
        if state.0 == 0
            && let Some(prev) = state.1.take()
        {
            std::panic::set_hook(prev);
        }
    }
}

/// Run every grid point through the consumer's `MemberRunner` via the engine
/// `sweep` (disjoint parallel members, C1 enumeration order). Faults are
/// captured per slot (the closure must return a report), and the LOWEST
/// enumeration index's fault aborts the stage after the sweep joins —
/// deterministic regardless of thread completion order. The placeholder
/// report for a faulted slot is never persisted or ranked: a captured fault
/// returns before any family write.
fn run_members(
    cell: &CellSpec,
    grid: &SweepGrid,
    runner: &dyn MemberRunner,
) -> Result<SweepFamily, ExecFault> {
    let space = ListSpace::new(&grid.specs, grid.points.clone()).map_err(|e| {
        ExecFault::PipelineShape {
            detail: format!("axis grid does not form a valid param space: {e:?}"),
        }
    })?;
    // The engine closure receives only the point's cells; recover the
    // enumeration index by value (duplicate points — possible only when an
    // axis repeats a value — collapse to the first index, which is exactly
    // the lowest-index attribution this capture exists for).
    let cells: Vec<Vec<Cell>> =
        grid.points.iter().map(|p| p.iter().map(|s| s.cell()).collect()).collect();
    let faults: Mutex<Vec<(usize, MemberFault)>> = Mutex::new(Vec::new());
    let family = sweep(&space, |point: &[Cell]| {
        let params = zip_params(&grid.specs, point);
        let called = {
            let _silence = SilencedPanic::enter();
            std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
                runner.run_member(cell, &params, cell.window_ms)
            }))
        };
        match called {
            Ok(Ok(report)) => report,
            Ok(Err(fault)) => {
                let index = cells
                    .iter()
                    .position(|c| c.as_slice() == point)
                    .expect("the sweep only enumerates the grid's own points");
                faults.lock().expect("fault capture lock").push((index, fault));
                placeholder_report(&params, cell.window_ms)
            }
            Err(payload) => {
                let index = cells
                    .iter()
                    .position(|c| c.as_slice() == point)
                    .expect("the sweep only enumerates the grid's own points");
                faults
                    .lock()
                    .expect("fault capture lock")
                    .push((index, MemberFault::Panic(panic_message(payload))));
                placeholder_report(&params, cell.window_ms)
            }
        }
    });
    let mut captured = faults.into_inner().expect("fault capture lock");
    captured.sort_by_key(|&(index, _)| index);
    if let Some((_, fault)) = captured.into_iter().next() {
        return Err(ExecFault::Member(fault));
    }
    Ok(family)
}

/// #272: which executor faults are per-cell outcomes vs campaign-wide
/// refusals. Member and window faults are data-and-cell-local; everything else
/// (doc-shape, registry) stays a global refusal returned to the caller.
fn contain(fault: ExecFault, stage: usize) -> Result<aura_registry::CellFault, ExecFault> {
    match fault {
        ExecFault::Member(m) => Ok(aura_registry::CellFault {
            stage,
            kind: member_kind(&m),
            detail: member_fault_prose(&m),
        }),
        ExecFault::Window { stage: s, detail } => Ok(aura_registry::CellFault {
            stage: s,
            kind: aura_registry::CellFaultKind::Window,
            detail,
        }),
        other => Err(other),
    }
}

fn member_kind(m: &MemberFault) -> aura_registry::CellFaultKind {
    use aura_registry::CellFaultKind as K;
    match m {
        MemberFault::NoData { .. } => K::NoData,
        MemberFault::Bind(_) => K::Bind,
        MemberFault::Run(_) => K::Run,
        MemberFault::Panic(_) => K::Panic,
    }
}

/// Prose for a member fault. `pub` (re-exported at the crate root) so the
/// CLI's `exec_fault_prose` phrases the `ExecFault::Member` arm by calling
/// this directly instead of carrying its own duplicate of the wording.
/// Debug-leak-free, self-describing.
pub fn member_fault_prose(m: &MemberFault) -> String {
    match m {
        MemberFault::NoData { instrument, window_ms } => format!(
            "no data for instrument {instrument} in window [{}, {}] (epoch-ms)",
            window_ms.0, window_ms.1
        ),
        MemberFault::Bind(detail) => format!("a member failed to bind: {detail}"),
        MemberFault::Run(detail) => format!("a member failed to run: {detail}"),
        MemberFault::Panic(detail) => format!("a member panicked: {detail}"),
    }
}

/// Best-effort panic-payload message (String / &str payloads; else a fixed
/// marker). `AssertUnwindSafe` at the call sites is a documented judgement: the
/// runner is reused for later cells; the CLI runner holds read-only state.
fn panic_message(payload: Box<dyn std::any::Any + Send>) -> String {
    payload
        .downcast_ref::<String>()
        .cloned()
        .or_else(|| payload.downcast_ref::<&str>().map(|s| s.to_string()))
        .unwrap_or_else(|| "member panicked (non-string payload)".to_string())
}

/// Assemble the (CellOutcome, CellRealization) pair for a cell that failed at a
/// stage (#272): the realized stage prefix survives in `stages`, `families`/
/// `selections` carry whatever already-persisted stages produced (empty for a
/// sweep-first fault; the sweep family/selection for a later wf-stage fault),
/// no nominee, the fault stamped on both, coverage from the runner.
fn failed_cell(
    cell: &CellSpec,
    stages: Vec<StageRealization>,
    families: Vec<StageFamily>,
    selections: Vec<StageSelectionOut>,
    fault: aura_registry::CellFault,
    runner: &dyn MemberRunner,
) -> (CellOutcome, CellRealization) {
    (
        CellOutcome { families, selections, nominee: None },
        CellRealization {
            strategy: cell.strategy_id.clone(),
            instrument: cell.instrument.clone(),
            window_ms: cell.window_ms,
            stages,
            regime: cell.regime,
            regime_ordinal: cell.regime_ordinal,
            fault: Some(fault),
            coverage: runner.window_coverage(cell),
        },
    )
}

/// Sentinel `RunManifest::broker` value stamped onto [`placeholder_report`]'s
/// output — the single source both the setter and the survivor filter
/// (`run_members`'s `... != FAULTED_MEMBER_PLACEHOLDER_BROKER` sibling) read,
/// so the two can never drift apart into silently pooling a faulted
/// placeholder into a real family/OOS selection.
const FAULTED_MEMBER_PLACEHOLDER_BROKER: &str = "faulted-member-placeholder";

/// Slot filler for a faulted member: the engine sweep contract needs one
/// report per slot, but a captured fault aborts `execute` before any family
/// write, so this report is never persisted and never ranked.
fn placeholder_report(params: &[(String, Scalar)], window_ms: (i64, i64)) -> RunReport {
    RunReport {
        manifest: RunManifest {
            commit: String::new(),
            params: params.to_vec(),
            defaults: Vec::new(),
            window: (Timestamp(window_ms.0), Timestamp(window_ms.1)),
            seed: 0,
            broker: FAULTED_MEMBER_PLACEHOLDER_BROKER.to_string(),
            selection: None,
            instrument: None,
            topology_hash: None,
            project: None,
        },
        metrics: RunMetrics { total_pips: 0.0, max_drawdown: 0.0, bias_sign_flips: 0, r: None },
    }
}

/// The shared in-sample-winner dispatch: called both for the sweep stage's
/// own family and for each walk-forward window's in-sample family. deflate=
/// true composes only with argmax (preflighted: `DeflatePlateauConflict`),
/// seeding the deflation nulls from the CAMPAIGN seed (C1: a campaign's
/// realization is a pure function of the document). Bare argmax has no
/// annotation to compute, so its `FamilySelection` is synthesized annotation-
/// free; `raw_winner_metric` mirrors the registry's ranking read (an R metric
/// against `r: None` ranks `NEG_INFINITY`). The plateau arms (`PlateauMean`/
/// `PlateauWorst`) ignore `deflate` outright (the `_` guard): they never
/// deflate regardless of the caller's flag, so the walk-forward caller's
/// hardcoded `deflate=true` only ever takes effect on the `Argmax` arm.
fn select_sweep_winner(
    family: &SweepFamily,
    axis_lens: &[usize],
    metric: &str,
    select: SelectRule,
    deflate: bool,
    seed: u64,
) -> Result<(SweepPoint, FamilySelection), ExecFault> {
    match (select, deflate) {
        (SelectRule::Argmax, true) => {
            optimize_deflated(family, metric, DEFLATION_N_RESAMPLES, DEFLATION_BLOCK_LEN, seed)
                .map_err(ExecFault::Registry)
        }
        (SelectRule::Argmax, false) => {
            let winner = optimize(family, metric).map_err(ExecFault::Registry)?;
            let raw = member_metric(&winner.report, metric).unwrap_or(f64::NEG_INFINITY);
            let selection = FamilySelection {
                selection_metric: metric.to_string(),
                n_trials: family.points.len(),
                raw_winner_metric: raw,
                mode: SelectionMode::Argmax,
                deflated_score: None,
                overfit_probability: None,
                n_resamples: None,
                block_len: None,
                seed: None,
                neighbourhood_score: None,
                n_neighbours: None,
            };
            Ok((winner, selection))
        }
        (SelectRule::PlateauMean, _) => {
            optimize_plateau(family, axis_lens, metric, PlateauMode::Mean)
                .map_err(ExecFault::Registry)
        }
        (SelectRule::PlateauWorst, _) => {
            optimize_plateau(family, axis_lens, metric, PlateauMode::Worst)
                .map_err(ExecFault::Registry)
        }
    }
}

/// A family point's tag-free cells lifted back to self-describing scalars in
/// `space` slot order — the survivor-point form the walk-forward stage
/// re-sweeps through a `ListSpace`.
fn scalar_point(space: &[ParamSpec], point: &[Cell]) -> Vec<Scalar> {
    space.iter().zip(point).map(|(ps, c)| Scalar::from_cell(ps.kind, *c)).collect()
}

/// The pooled walk-forward OOS trade-R series: the family reports' `net_trade_rs`
/// concatenated in report order — which IS roll order (the wf stage builds its
/// family via `walkforward_member_reports_of`, per-window OOS reports in roll
/// order, over the surviving folds). Reports without an R block contribute
/// nothing.
fn pooled_net_trade_rs(reports: &[RunReport]) -> Vec<f64> {
    let mut pooled = Vec::new();
    for report in reports {
        if let Some(r) = &report.metrics.r {
            pooled.extend_from_slice(&r.net_trade_rs);
        }
    }
    pooled
}

/// A placeholder for a faulted window slot: arity-correct chosen params (the
/// engine asserts every window's chosen-point arity against the param-space),
/// empty OOS equity, inert report. Never surfaces — a recorded window fault
/// fails the stage after the roll, before any registry write.
fn placeholder_window_run(specs: &[ParamSpec]) -> WindowRun {
    WindowRun {
        chosen_params: vec![Cell::from_i64(0); specs.len()],
        oos_equity: vec![],
        oos_report: placeholder_report(&[], (0, 0)),
    }
}

/// Execute one `std::walk_forward` stage over the cell's surviving points:
/// roll (IS, OOS) window splits over the cell window — entirely in ms
/// (`Timestamp` is unit-agnostic `i64`; the driver owns ms->ns at its data
/// seam) — then per window sweep the survivor points over the IS bounds and
/// pick the winner via `select_sweep_winner` (the same dispatch the sweep
/// stage uses), seeded from the campaign seed: `Argmax` keeps the shipped
/// trials-deflation provenance, and `PlateauMean`/`PlateauWorst` read
/// `axis_lens` to score the closed neighbourhood — sound only because
/// preflight refuses plateau whenever a gate precedes this stage (a gate
/// filters survivors below the full grid, breaking the lattice `axis_lens`
/// assumes intact). Then run the winner over the OOS bounds with the
/// selection stamped on its fresh report, and append the per-window OOS
/// reports as a `FamilyKind::WalkForward` family.
#[allow(clippy::too_many_arguments)]
fn run_walk_forward_stage(
    seed: u64,
    cell: &CellSpec,
    stage: usize,
    block: &StageBlock,
    specs: &[ParamSpec],
    axis_lens: &[usize],
    survivors: &[Vec<Scalar>],
    family_name: &str,
    runner: &dyn MemberRunner,
    registry: &Registry,
) -> Result<WfStageOutcome, ExecFault> {
    let StageBlock::WalkForward { in_sample_ms, out_of_sample_ms, step_ms, mode, metric, select } =
        block
    else {
        return Err(ExecFault::PipelineShape {
            detail: format!("stage {stage} is not std::walk_forward"),
        });
    };
    if survivors.is_empty() {
        // execute() truncates a cell at an empty gate before reaching here;
        // refuse (never panic) if that contract is ever violated.
        return Err(ExecFault::PipelineShape {
            detail: format!("stage {stage}: walk_forward reached with zero survivor points"),
        });
    }
    // Preflight guarantees the u64 lengths fit i64.
    let span = (Timestamp(cell.window_ms.0), Timestamp(cell.window_ms.1));
    let roll_mode = match mode {
        WfMode::Rolling => RollMode::Rolling,
        WfMode::Anchored => RollMode::Anchored,
    };
    let roller = WindowRoller::new(
        span,
        *in_sample_ms as i64,
        *out_of_sample_ms as i64,
        *step_ms as i64,
        roll_mode,
    )
    .map_err(|e| ExecFault::Window {
        stage,
        // The detail string reaches user-facing seams verbatim — phrase both
        // roller refusals in the doc's ms unit, never a Debug form.
        detail: match e {
            WalkForwardError::SpanTooShort { .. } => format!(
                "in_sample_ms + out_of_sample_ms = {} ms exceeds the campaign window [{}, {}] ({} ms)",
                in_sample_ms + out_of_sample_ms,
                cell.window_ms.0,
                cell.window_ms.1,
                cell.window_ms.1 - cell.window_ms.0 + 1,
            ),
            // Zero lengths are doc-tier faults (preflight); phrased anyway.
            WalkForwardError::NonPositiveLength { field, value } => {
                format!("walk_forward {field} must be > 0 (got {value})")
            }
        },
    })?;
    // A second identical roller enumerates the bounds up front: the engine's
    // walk_forward consumes its roller, and the parallel closure needs each
    // window's roll index for deterministic fault attribution.
    let bounds: Vec<WindowBounds> = WindowRoller::new(
        span,
        *in_sample_ms as i64,
        *out_of_sample_ms as i64,
        *step_ms as i64,
        roll_mode,
    )
    .expect("identical roller config validated above")
    .collect();

    let is_space = ListSpace::new(specs, survivors.to_vec()).map_err(|e| {
        ExecFault::PipelineShape {
            detail: format!("stage {stage}: survivor points do not fit the param space: {e:?}"),
        }
    })?;
    let is_points = is_space.points();

    // Windows run in parallel and the engine closure cannot return Err:
    // faulted windows record here (keyed by roll index) and yield a
    // placeholder run; after the roll the lowest-index fault wins
    // (deterministic prose, the sweep-stage capture pattern one level up).
    let faults: Mutex<Vec<(usize, ExecFault)>> = Mutex::new(Vec::new());

    let result = walk_forward(roller, specs.to_vec(), |w| {
        let widx =
            bounds.iter().position(|b| *b == w).expect("bounds come from an identical roller");
        // IS: engine sweep of the survivor points over the in-sample bounds,
        // with the same member fault capture the sweep stage uses.
        let is_faults: Mutex<Vec<(usize, MemberFault)>> = Mutex::new(Vec::new());
        let is_family = sweep(&is_space, |cells| {
            let params = zip_params(specs, cells);
            let called = {
                let _silence = SilencedPanic::enter();
                std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
                    runner.run_member(cell, &params, (w.is.0 .0, w.is.1 .0))
                }))
            };
            match called {
                Ok(Ok(report)) => report,
                Ok(Err(fault)) => {
                    let midx = is_points
                        .iter()
                        .position(|p| p.as_slice() == cells)
                        .expect("sweep enumerates exactly is_points");
                    is_faults.lock().unwrap().push((midx, fault));
                    placeholder_report(&[], (0, 0))
                }
                Err(payload) => {
                    let midx = is_points
                        .iter()
                        .position(|p| p.as_slice() == cells)
                        .expect("sweep enumerates exactly is_points");
                    is_faults.lock().unwrap().push((midx, MemberFault::Panic(panic_message(payload))));
                    placeholder_report(&[], (0, 0))
                }
            }
        });
        let member_faults = is_faults.into_inner().expect("no thread panicked holding the lock");
        if let Some((_, fault)) = member_faults.into_iter().min_by_key(|(i, _)| *i) {
            faults.lock().unwrap().push((widx, ExecFault::Member(fault)));
            return placeholder_window_run(specs);
        }
        // IS winner: the same select_sweep_winner dispatch the sweep stage
        // uses, seeded from the campaign seed. deflate=true keeps Argmax's
        // shipped trials-deflation provenance identical to before this
        // dispatch existed; the Plateau arms ignore the deflate flag.
        let (winner, selection) =
            match select_sweep_winner(&is_family, axis_lens, metric, *select, true, seed) {
                Ok(picked) => picked,
                Err(e) => {
                    faults.lock().unwrap().push((widx, e));
                    return placeholder_window_run(specs);
                }
            };
        // OOS: run the winner over the out-of-sample bounds; the selection is
        // stamped on the fresh OOS report (a new record, never a mutation of
        // a stored one).
        let oos_params = zip_params(specs, &winner.params);
        let oos_called = {
            let _silence = SilencedPanic::enter();
            std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
                runner.run_member(cell, &oos_params, (w.oos.0 .0, w.oos.1 .0))
            }))
        };
        let mut oos_report = match oos_called {
            Ok(Ok(report)) => report,
            Ok(Err(fault)) => {
                faults.lock().unwrap().push((widx, ExecFault::Member(fault)));
                return placeholder_window_run(specs);
            }
            Err(payload) => {
                faults
                    .lock()
                    .unwrap()
                    .push((widx, ExecFault::Member(MemberFault::Panic(panic_message(payload)))));
                return placeholder_window_run(specs);
            }
        };
        oos_report.manifest.selection = Some(selection);
        WindowRun { chosen_params: winner.params, oos_equity: vec![], oos_report }
    });

    // Registry faults (optimizer/persistence broken) stay a global abort;
    // Member/Window faults become per-fold WindowFaults via the same
    // `contain` mapping the cell-level containment uses. One drain, in
    // ascending window-index order — ExecFault/RegistryError are not Clone,
    // so this is a single ownership-respecting pass rather than iter()-then-
    // into_iter(); the first Registry fault encountered (ascending order =
    // lowest index) wins, matching the existing lowest-index-fault convention.
    let mut captured = faults.into_inner().expect("no thread panicked holding the lock");
    captured.sort_by_key(|(idx, _)| *idx);
    let mut window_faults: Vec<aura_registry::WindowFault> = Vec::with_capacity(captured.len());
    for (widx, fault) in captured {
        match contain(fault, stage) {
            Ok(cf) => window_faults.push(aura_registry::WindowFault {
                window_ordinal: widx,
                kind: cf.kind,
                detail: cf.detail,
            }),
            Err(global) => return Err(global),
        }
    }

    // Keep only real OOS reports (drop faulted-fold placeholders by the broker
    // sentinel) before building the family.
    let surviving: Vec<&aura_engine::WindowOutcome> = result
        .windows
        .iter()
        .filter(|w| w.run.oos_report.manifest.broker != FAULTED_MEMBER_PLACEHOLDER_BROKER)
        .collect();
    if surviving.is_empty() {
        // Every fold failed — the cell fails at this stage.
        let first = window_faults.first().cloned();
        return Ok(WfStageOutcome::CellFailed(aura_registry::CellFault {
            stage,
            kind: first.as_ref().map(|w| w.kind).unwrap_or(aura_registry::CellFaultKind::Run),
            detail: format!(
                "all {} walk_forward folds failed{}",
                result.windows.len(),
                first.map(|w| format!("; first: {}", w.detail)).unwrap_or_default()
            ),
        }));
    }
    let reports = walkforward_member_reports_of(&surviving);
    let family_id = registry
        .append_family(family_name, FamilyKind::WalkForward, &reports)
        .map_err(ExecFault::Registry)?;
    Ok(WfStageOutcome::Ran {
        family: StageFamily { stage, block: "std::walk_forward", family_id, reports },
        window_faults,
    })
}

/// #272: a wf stage either ran (its family + any per-fold holes) or the whole
/// cell failed (every fold faulted). A Registry fault is a separate global Err
/// (returned directly from `run_walk_forward_stage`, never wrapped here).
enum WfStageOutcome {
    Ran { family: StageFamily, window_faults: Vec<aura_registry::WindowFault> },
    CellFailed(aura_registry::CellFault),
}

/// The surviving (non-faulted) walk-forward windows' OOS reports, in roll
/// order — the same mapping `walkforward_member_reports` does over a full
/// result, applied to the filtered survivor slice.
fn walkforward_member_reports_of(windows: &[&aura_engine::WindowOutcome]) -> Vec<RunReport> {
    windows.iter().map(|w| w.run.oos_report.clone()).collect()
}