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, walkforward_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 })
}

/// 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: (ordinal into the nearest preceding
    // family_id-bearing stage's family, param point, member report).
    let mut survivors: Vec<(usize, Vec<Scalar>, RunReport)> = 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::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 = run_members(cell, &grid, 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,
                        });
                    }
                    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,
                        });
                    }
                }
                // The whole family flows on (decision 3: `select` names the
                // recorded selection, never a filter).
                survivors = 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`).
                survivors.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> = survivors.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,
                });
                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>> =
                    survivors.iter().map(|(_, point, _)| point.clone()).collect();
                let fam = run_walk_forward_stage(
                    seed,
                    cell,
                    stage_ordinal,
                    stage,
                    &grid.specs,
                    &grid.axis_lens,
                    &survivor_points,
                    &family_name,
                    runner,
                    registry,
                )?;
                stages.push(StageRealization {
                    block: "std::walk_forward".to_string(),
                    family_id: Some(fam.family_id.clone()),
                    survivor_ordinals: None,
                    selection: None,
                    bootstrap: None,
                });
                // Nominate the LAST window's OOS report (roll order per
                // walkforward_member_reports) 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);
            }
            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 => StageBootstrap::PerSurvivor(
                            survivors
                                .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),
                });
            }
            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,
            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 }
}

/// 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);
        match runner.run_member(cell, &params, cell.window_ms) {
            Ok(report) => report,
            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)
            }
        }
    });
    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)
}

/// 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".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`, per-window OOS reports in roll
/// order). 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<StageFamily, 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| {
            match runner.run_member(cell, &zip_params(specs, cells), (w.is.0 .0, w.is.1 .0)) {
                Ok(report) => report,
                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))
                }
            }
        });
        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 mut oos_report = match runner.run_member(
            cell,
            &zip_params(specs, &winner.params),
            (w.oos.0 .0, w.oos.1 .0),
        ) {
            Ok(report) => report,
            Err(fault) => {
                faults.lock().unwrap().push((widx, ExecFault::Member(fault)));
                return placeholder_window_run(specs);
            }
        };
        oos_report.manifest.selection = Some(selection);
        WindowRun { chosen_params: winner.params, oos_equity: vec![], oos_report }
    });

    let window_faults = faults.into_inner().expect("no thread panicked holding the lock");
    if let Some((_, fault)) = window_faults.into_iter().min_by_key(|(i, _)| *i) {
        return Err(fault);
    }

    let reports = walkforward_member_reports(&result);
    let family_id = registry
        .append_family(family_name, FamilyKind::WalkForward, &reports)
        .map_err(ExecFault::Registry)?;
    Ok(StageFamily { stage, block: "std::walk_forward", family_id, reports })
}