Aura/crates/aura-campaign/tests/execute.rs

//! Executor semantics over a fake `MemberRunner` (hermetic — no engine
//! harness, no data): cell loop, sweep stage, gate filtering, zero-survivor
//! truncation, determinism, fault attribution, deflation seeding.

use std::collections::BTreeMap;

use aura_campaign::{
    execute, CellSpec, ExecFault, MemberFault, MemberRunner, DEFAULT_PARALLEL_INSTRUMENTS,
};
use aura_core::{Scalar, ScalarKind, Timestamp};
use aura_engine::{r_bootstrap, RMetrics, RunManifest, RunMetrics, RunReport, SelectionMode};
use aura_registry::{generalization, FamilyKind, Registry, RegistryError, StageBootstrap};
use aura_research::{
    Axis, CampaignDoc, Cmp, DataSection, DocKind, DocRef, Predicate, Presentation, ProcessDoc,
    ProcessRef, SelectRule, StageBlock, StrategyEntry, SweepSelection, WfMode, Window,
};

const CAMPAIGN_ID: &str = "aaaabbbbccccddddeeeeffff0000111122223333444455556666777788889999";
const PROCESS_ID: &str = "9999888877776666555544443333222211110000ffffeeeeddddccccbbbbaaaa";
const STRATEGY_ID: &str = "1111222233334444555566667777888899990000aaaabbbbccccddddeeeeffff";

/// Deterministic fake: every planted metric is a pure function of
/// (params, instrument). `fast`/`slow` are the two I64 axes;
/// net = (fast*10 + slow)/100, negated for the instrument named "AAA".
struct FakeRunner {
    /// (params that fault, the fault) — checked before planting a report.
    faults: Vec<(Vec<(String, Scalar)>, MemberFault)>,
}

impl FakeRunner {
    fn clean() -> Self {
        FakeRunner { faults: Vec::new() }
    }
}

fn param_i64(params: &[(String, Scalar)], name: &str) -> i64 {
    params
        .iter()
        .find(|(n, _)| n == name)
        .map(|(_, v)| v.as_i64())
        .expect("planted axis present")
}

/// The deterministic 4-trade R series a planted report carries (matches
/// `n_trades: 4`; non-constant so block resampling has structure). In-memory
/// only: `net_trade_rs` is serde-skipped and excluded from `PartialEq`, so every
/// existing equality and round-trip assertion stays green.
fn planted_net_trade_rs(net: f64) -> Vec<f64> {
    vec![net, -net, 2.0 * net, net]
}

fn planted_report(cell: &CellSpec, params: &[(String, Scalar)], window_ms: (i64, i64)) -> RunReport {
    let fast = param_i64(params, "fast");
    let slow = param_i64(params, "slow");
    let mut net = (fast * 10 + slow) as f64 / 100.0;
    if cell.instrument == "AAA" {
        net = -net;
    }
    RunReport {
        manifest: RunManifest {
            commit: "fake".to_string(),
            params: params.to_vec(),
            defaults: Vec::new(),
            window: (Timestamp(window_ms.0), Timestamp(window_ms.1)),
            seed: 0,
            broker: "fake".to_string(),
            selection: None,
            instrument: Some(cell.instrument.clone()),
            topology_hash: Some(cell.strategy_id.clone()),
            project: None,
        },
        metrics: RunMetrics {
            total_pips: net * 100.0,
            max_drawdown: 1.0,
            bias_sign_flips: 1,
            r: Some(RMetrics {
                expectancy_r: net,
                n_trades: 4,
                win_rate: 0.5,
                avg_win_r: 1.0,
                avg_loss_r: -0.5,
                profit_factor: 2.0,
                max_r_drawdown: 0.5,
                n_open_at_end: 0,
                sqn: net,
                sqn_normalized: net,
                net_expectancy_r: net,
                conviction_terciles_r: [0.0, 0.0, 0.0],
                net_trade_rs: planted_net_trade_rs(net),
            }),
        },
    }
}

impl MemberRunner for FakeRunner {
    fn run_member(
        &self,
        cell: &CellSpec,
        params: &[(String, Scalar)],
        window_ms: (i64, i64),
    ) -> Result<RunReport, MemberFault> {
        if let Some((_, fault)) = self.faults.iter().find(|(p, _)| p.as_slice() == params) {
            return Err(fault.clone());
        }
        Ok(planted_report(cell, params, window_ms))
    }
}

/// Wraps the clean fake and strips the R block from ONE param point's report —
/// the zero-trade survivor for the mc degenerate pin.
struct ZeroTradeRunner {
    inner: FakeRunner,
    strip: Vec<(String, Scalar)>,
}

impl MemberRunner for ZeroTradeRunner {
    fn run_member(
        &self,
        cell: &CellSpec,
        params: &[(String, Scalar)],
        window_ms: (i64, i64),
    ) -> Result<RunReport, MemberFault> {
        let mut report = self.inner.run_member(cell, params, window_ms)?;
        if params == self.strip.as_slice() {
            report.metrics.r = None;
        }
        Ok(report)
    }
}

/// A runner whose `run_member` always panics — the #272 panic-containment
/// fixture (`catch_unwind` at the member boundary must turn this into a
/// `MemberFault::Panic`, never an unwind out of `execute`).
struct PanicRunner;

impl MemberRunner for PanicRunner {
    fn run_member(
        &self,
        _cell: &CellSpec,
        _params: &[(String, Scalar)],
        _window_ms: (i64, i64),
    ) -> Result<RunReport, MemberFault> {
        panic!("planted member panic")
    }
}

/// fast in {2,3}, slow in {6,9} -> 4 odometer points (last axis fastest):
/// 0:(2,6) net .26 | 1:(2,9) net .29 | 2:(3,6) net .36 | 3:(3,9) net .39.
fn axes_2x2() -> BTreeMap<String, Axis> {
    let mut axes = BTreeMap::new();
    axes.insert(
        "fast".to_string(),
        Axis { kind: ScalarKind::I64, values: vec![Scalar::i64(2), Scalar::i64(3)] },
    );
    axes.insert(
        "slow".to_string(),
        Axis { kind: ScalarKind::I64, values: vec![Scalar::i64(6), Scalar::i64(9)] },
    );
    axes
}

fn campaign(instruments: &[&str]) -> CampaignDoc {
    CampaignDoc {
        format_version: 1,
        kind: DocKind::Campaign,
        name: "exec-test".to_string(),
        description: None,
        data: DataSection {
            instruments: instruments.iter().map(|s| s.to_string()).collect(),
            windows: vec![Window { from_ms: 1_000, to_ms: 9_000 }],
            bindings: BTreeMap::new(),
        },
        risk: vec![],
        cost: vec![],
        strategies: vec![StrategyEntry {
            r#ref: DocRef::ContentId(STRATEGY_ID.to_string()),
            axes: axes_2x2(),
        }],
        process: ProcessRef { r#ref: DocRef::ContentId(PROCESS_ID.to_string()) },
        seed: 7,
        presentation: Presentation { persist_taps: vec![], emit: vec![] },
    }
}

fn sweep_stage(deflate: bool) -> StageBlock {
    StageBlock::Sweep {
        selection: Some(SweepSelection {
            metric: "net_expectancy_r".to_string(),
            select: SelectRule::Argmax,
            deflate,
        }),
    }
}

fn selection_free_sweep_stage() -> StageBlock {
    StageBlock::Sweep { selection: None }
}

fn gate_stage(value: f64) -> StageBlock {
    StageBlock::Gate {
        all: vec![Predicate { metric: "net_expectancy_r".to_string(), cmp: Cmp::Gt, value }],
    }
}

/// 2 rolling windows over the campaign window [1_000, 9_000]: IS 4_000 ms,
/// OOS 2_000 ms, step 2_000 ms -> OOS [5_000, 6_999] and [7_000, 8_999]
/// (window 2 would need OOS end 10_999 > 9_000, so the roller stops at 2).
fn wf_stage() -> StageBlock {
    StageBlock::WalkForward {
        in_sample_ms: 4_000,
        out_of_sample_ms: 2_000,
        step_ms: 2_000,
        mode: WfMode::Rolling,
        metric: "net_expectancy_r".to_string(),
        select: SelectRule::Argmax,
    }
}

fn generalize_stage() -> StageBlock {
    StageBlock::Generalize { metric: "net_expectancy_r".to_string() }
}

/// A registry whose family store path is pre-occupied by a directory:
/// `append_family`'s `OpenOptions::open` then fails with a real IO error
/// (`ErrorKind::IsADirectory`) — a `RegistryError::Io` that is neither a
/// `MemberFault` nor a `WindowFault`, so `run_cell`'s `?` on `append_family`
/// returns it as a non-containable `ExecFault` (never routed through
/// `contain`), unlike the `FakeRunner`-planted member faults used elsewhere
/// in this file.
fn temp_registry_with_blocked_family_store(name: &str) -> Registry {
    let dir = std::path::Path::new(env!("CARGO_TARGET_TMPDIR"))
        .join(format!("aura-campaign-exec-{name}"));
    let _ = std::fs::remove_dir_all(&dir);
    std::fs::create_dir_all(&dir).expect("create temp registry dir");
    std::fs::create_dir_all(dir.join("families.jsonl")).expect("occupy the family store path");
    Registry::open(dir.join("runs.jsonl"))
}

fn process(pipeline: Vec<StageBlock>) -> ProcessDoc {
    ProcessDoc {
        format_version: 1,
        kind: DocKind::Process,
        name: "exec-test-process".to_string(),
        description: None,
        pipeline,
    }
}

fn strategies() -> Vec<(String, String)> {
    vec![(STRATEGY_ID.to_string(), r#"{"format_version":1}"#.to_string())]
}

/// Per-test registry DIRECTORY (family/campaign stores are per-directory
/// siblings of the runs path — the aura-registry temp-dir idiom).
fn temp_registry(name: &str) -> Registry {
    let dir = std::path::Path::new(env!("CARGO_TARGET_TMPDIR"))
        .join(format!("aura-campaign-exec-{name}"));
    let _ = std::fs::remove_dir_all(&dir);
    std::fs::create_dir_all(&dir).expect("create temp registry dir");
    Registry::open(dir.join("runs.jsonl"))
}

#[test]
fn execute_sweep_only_records_family_and_selection() {
    let reg = temp_registry("sweep_only");
    let doc = campaign(&["EURUSD"]);
    let proc_doc = process(vec![sweep_stage(false)]);
    let out = execute(CAMPAIGN_ID, &doc, &proc_doc, &strategies(), &FakeRunner::clean(), &reg, DEFAULT_PARALLEL_INSTRUMENTS)
        .expect("sweep-only campaign executes");

    // outcome payloads: one cell, one family of 4 members, one selection
    assert_eq!(out.run, 0);
    assert_eq!(out.cells.len(), 1);
    assert_eq!(out.cells[0].families.len(), 1);
    assert_eq!(out.cells[0].families[0].reports.len(), 4);
    let expected_family_id = format!("{}-0-EURUSD-w0-r0-s0-0", &CAMPAIGN_ID[..8]);
    assert_eq!(out.cells[0].families[0].family_id, expected_family_id);

    // planted argmax: (3,9) is odometer point 3 (last axis fastest)
    assert_eq!(out.cells[0].selections.len(), 1);
    let sel = &out.cells[0].selections[0];
    assert_eq!(sel.winner_ordinal, 3);
    assert_eq!(
        sel.params,
        vec![("fast".to_string(), Scalar::i64(3)), ("slow".to_string(), Scalar::i64(9))],
    );

    // family persisted: 4 sweep members under the derived id, ordinal-ordered
    let members = reg.load_family_members().expect("load members");
    assert_eq!(members.len(), 4);
    assert!(members.iter().all(|m| m.kind == FamilyKind::Sweep));
    assert!(members.iter().all(|m| m.family_id() == expected_family_id));
    assert_eq!(members.iter().map(|m| m.ordinal).collect::<Vec<_>>(), vec![0, 1, 2, 3]);

    // realization persisted: one record linking the family + the selection
    let runs = reg.load_campaign_runs().expect("load campaign runs");
    assert_eq!(runs.len(), 1);
    assert_eq!(runs[0], out.record);
    assert_eq!(runs[0].campaign, CAMPAIGN_ID);
    assert_eq!(runs[0].process, PROCESS_ID);
    assert_eq!(runs[0].run, 0);
    assert_eq!(runs[0].seed, 7);
    assert_eq!(
        runs[0].trace_name, None,
        "empty persist_taps claims no trace name (0109 contract)"
    );
    assert_eq!(runs[0].cells.len(), 1);
    let cell = &runs[0].cells[0];
    assert_eq!(cell.strategy, STRATEGY_ID);
    assert_eq!(cell.instrument, "EURUSD");
    assert_eq!(cell.window_ms, (1_000, 9_000));
    assert_eq!(cell.stages.len(), 1);
    assert_eq!(cell.stages[0].block, "std::sweep");
    assert_eq!(cell.stages[0].family_id.as_deref(), Some(expected_family_id.as_str()));
    let stored = cell.stages[0].selection.as_ref().expect("sweep stage carries a selection");
    assert_eq!(stored.winner_ordinal, 3);
    assert_eq!(stored.selection.mode, SelectionMode::Argmax);
    assert_eq!(stored.selection.n_trials, 4);
    assert!((stored.selection.raw_winner_metric - 0.39).abs() < 1e-12);
    assert_eq!(stored.selection.seed, None, "no deflation annotation without deflate");
}

/// A selection-free sweep records the family (the whole grid, exactly as a
/// selected sweep does) but never a selection or a nominee: there is no
/// winner to name, and nothing downstream to nominate for.
#[test]
fn execute_selection_free_sweep_records_family_without_selection() {
    let reg = temp_registry("selection_free_sweep");
    let doc = campaign(&["EURUSD"]);
    let proc_doc = process(vec![selection_free_sweep_stage()]);
    let out = execute(CAMPAIGN_ID, &doc, &proc_doc, &strategies(), &FakeRunner::clean(), &reg, DEFAULT_PARALLEL_INSTRUMENTS)
        .expect("selection-free sweep campaign executes");

    // outcome payloads: one cell, one family of 4 members, NO selection
    assert_eq!(out.cells.len(), 1);
    assert_eq!(out.cells[0].families.len(), 1);
    assert_eq!(out.cells[0].families[0].reports.len(), 4);
    assert_eq!(out.cells[0].families[0].block, "std::sweep");
    let expected_family_id = format!("{}-0-EURUSD-w0-r0-s0-0", &CAMPAIGN_ID[..8]);
    assert_eq!(out.cells[0].families[0].family_id, expected_family_id);
    assert!(out.cells[0].selections.is_empty(), "no winner is named without a selection group");

    // family persisted: 4 sweep members under the derived id, same as a
    // selected sweep — the selection-free path differs only in what it
    // records ABOUT the family, never in the family itself.
    let members = reg.load_family_members().expect("load members");
    assert_eq!(members.len(), 4);
    assert!(members.iter().all(|m| m.kind == FamilyKind::Sweep));

    // realization: the stage carries a family_id but no selection; no
    // nominee-dependent stage ran, so no generalization is recorded either.
    let runs = reg.load_campaign_runs().expect("load campaign runs");
    assert_eq!(runs.len(), 1);
    assert_eq!(runs[0], out.record);
    assert_eq!(runs[0].cells[0].stages.len(), 1);
    let stage = &runs[0].cells[0].stages[0];
    assert_eq!(stage.block, "std::sweep");
    assert_eq!(stage.family_id.as_deref(), Some(expected_family_id.as_str()));
    assert!(stage.selection.is_none(), "a selection-free sweep records no selection");
    assert!(runs[0].generalizations.is_empty());
}

#[test]
fn execute_gate_filters_per_member() {
    let reg = temp_registry("gate_filters");
    let doc = campaign(&["EURUSD"]);
    let proc_doc = process(vec![sweep_stage(false), gate_stage(0.3)]);
    let out = execute(CAMPAIGN_ID, &doc, &proc_doc, &strategies(), &FakeRunner::clean(), &reg, DEFAULT_PARALLEL_INSTRUMENTS)
        .expect("gated campaign executes");

    // planted nets 0.26 / 0.29 / 0.36 / 0.39 -> gt 0.3 keeps ordinals 2 and 3
    let cell = &out.record.cells[0];
    assert_eq!(cell.stages.len(), 2);
    assert_eq!(cell.stages[1].block, "std::gate");
    assert_eq!(cell.stages[1].survivor_ordinals, Some(vec![2, 3]));
    assert_eq!(cell.stages[1].family_id, None);
    assert_eq!(cell.stages[1].selection, None);
}

#[test]
fn execute_zero_survivors_truncates_cell_and_continues() {
    let reg = temp_registry("zero_survivors");
    // two instruments: the fake plants NEGATIVE nets for "AAA", positive for "BBB"
    let doc = campaign(&["AAA", "BBB"]);
    // sweep -> gate(net > 0) -> gate(net > -1000): AAA dies at stage 1, BBB passes both
    let proc_doc = process(vec![sweep_stage(false), gate_stage(0.0), gate_stage(-1000.0)]);
    let out = execute(CAMPAIGN_ID, &doc, &proc_doc, &strategies(), &FakeRunner::clean(), &reg, DEFAULT_PARALLEL_INSTRUMENTS)
        .expect("a zero-survivor cell is a valid result, not a fault");

    assert_eq!(out.record.cells.len(), 2, "the second cell still runs");
    // cell 0 (AAA): realization truncated AT the empty gate — stage 2 never realized
    let aaa = &out.record.cells[0];
    assert_eq!(aaa.instrument, "AAA");
    assert_eq!(aaa.stages.len(), 2);
    assert_eq!(aaa.stages[1].survivor_ordinals, Some(vec![]));
    // cell 1 (BBB): full pipeline realized
    let bbb = &out.record.cells[1];
    assert_eq!(bbb.instrument, "BBB");
    assert_eq!(bbb.stages.len(), 3);
    assert_eq!(bbb.stages[1].survivor_ordinals, Some(vec![0, 1, 2, 3]));
    assert_eq!(bbb.stages[2].survivor_ordinals, Some(vec![0, 1, 2, 3]));
}

/// #277: with `parallel_instruments = 1` (forcing 2 single-instrument
/// chunks) and 2 strategies x 2 instruments, the instrument-major chunk walk
/// executes cell (s0,AAA) then (s1,AAA) before either BBB cell — an
/// execution order that diverges from document order. The rebuild step must
/// still hand back cells in document order (C1): strategy-major, then
/// instrument, exactly as the pre-#277 nested loop did.
#[test]
fn execute_preserves_document_order_across_chunk_boundary() {
    let reg = temp_registry("chunk_order");
    let mut doc = campaign(&["AAA", "BBB"]);
    doc.strategies.push(doc.strategies[0].clone());
    let mut strats = strategies();
    strats.push(strats[0].clone());
    let proc_doc = process(vec![sweep_stage(false)]);
    let k = std::num::NonZeroUsize::new(1).expect("1 is nonzero");
    let out = execute(CAMPAIGN_ID, &doc, &proc_doc, &strats, &FakeRunner::clean(), &reg, k)
        .expect("chunked campaign executes");

    assert_eq!(out.cells.len(), 4, "2 strategies x 2 instruments x 1 window");
    let base = &CAMPAIGN_ID[..8];
    let family_id = |i: usize| out.record.cells[i].stages[0].family_id.clone();
    assert_eq!(family_id(0), Some(format!("{base}-0-AAA-w0-r0-s0-0")));
    assert_eq!(family_id(1), Some(format!("{base}-0-BBB-w0-r0-s0-0")));
    assert_eq!(family_id(2), Some(format!("{base}-1-AAA-w0-r0-s0-0")));
    assert_eq!(family_id(3), Some(format!("{base}-1-BBB-w0-r0-s0-0")));
    assert_eq!(
        out.record.cells.iter().map(|c| c.instrument.as_str()).collect::<Vec<_>>(),
        vec!["AAA", "BBB", "AAA", "BBB"],
        "document order (strategy-major, then instrument) survives the K=1 chunk walk",
    );
}

#[test]
fn execute_is_deterministic_twice() {
    let reg = temp_registry("deterministic_twice");
    let doc = campaign(&["EURUSD"]);
    let proc_doc = process(vec![sweep_stage(false)]);
    let first = execute(CAMPAIGN_ID, &doc, &proc_doc, &strategies(), &FakeRunner::clean(), &reg, DEFAULT_PARALLEL_INSTRUMENTS)
        .expect("first run");
    let second = execute(CAMPAIGN_ID, &doc, &proc_doc, &strategies(), &FakeRunner::clean(), &reg, DEFAULT_PARALLEL_INSTRUMENTS)
        .expect("second run");

    // run counter advances per campaign id
    assert_eq!((first.run, second.run), (0, 1));
    assert_eq!((first.record.run, second.record.run), (0, 1));

    // family ids differ ONLY by the "-{run}" suffix
    let base = format!("{}-0-EURUSD-w0-r0-s0", &CAMPAIGN_ID[..8]);
    assert_eq!(first.cells[0].families[0].family_id, format!("{base}-0"));
    assert_eq!(second.cells[0].families[0].family_id, format!("{base}-1"));

    // everything else in the record is identical (C1): normalize the two
    // divergent fields and compare whole records
    let mut normalized = second.record.clone();
    normalized.run = first.record.run;
    normalized.cells[0].stages[0].family_id = first.record.cells[0].stages[0].family_id.clone();
    assert_eq!(normalized, first.record);

    // both records stored, in order
    let runs = reg.load_campaign_runs().expect("load campaign runs");
    assert_eq!(runs.len(), 2);
    assert_eq!(runs[0], first.record);
    assert_eq!(runs[1], second.record);
}

/// #272: a member fault is contained as a failed cell, not a global abort —
/// the campaign run record still persists (the whole point of containment),
/// while a sweep-faulted cell writes no family (`run_members` returns before
/// `append_family`).
#[test]
fn execute_member_fault_records_a_failed_cell_and_continues() {
    let reg = temp_registry("member_fault");
    // two instruments; the planted fault is keyed by param point, so both
    // cells' sweeps hit it and both fail — the containment property under test
    // is that execute returns Ok and records every cell, not a global abort.
    let doc = campaign(&["AAA", "BBB"]);
    let proc_doc = process(vec![sweep_stage(false)]);
    let point = |fast: i64, slow: i64| {
        vec![("fast".to_string(), Scalar::i64(fast)), ("slow".to_string(), Scalar::i64(slow))]
    };
    // FakeRunner faults are matched by param point regardless of instrument;
    // plant one on the grid so AAA's (and BBB's) sweep hit it — the whole
    // sweep-faulted cell fails (grid hole compromises selection).
    let runner = FakeRunner {
        faults: vec![(point(2, 9), MemberFault::Run("boom".to_string()))],
    };
    let out = execute(CAMPAIGN_ID, &doc, &proc_doc, &strategies(), &runner, &reg, DEFAULT_PARALLEL_INSTRUMENTS)
        .expect("a faulted member is recorded, not a global abort");
    assert_eq!(out.record.cells.len(), 2, "both cells are recorded");
    for cell in &out.record.cells {
        let fault = cell.fault.as_ref().expect("sweep-fault fails the cell");
        assert_eq!(fault.stage, 0);
        assert_eq!(fault.kind, aura_registry::CellFaultKind::Run);
    }
    // The run record persisted (the whole point of containment); a sweep-fault
    // cell writes no family (run_members returns before append_family).
    assert_eq!(reg.load_campaign_runs().expect("load runs").len(), 1);
    assert!(reg.load_family_members().expect("load members").is_empty());
}

/// #272: a panic that unwinds out of `MemberRunner::run_member` is caught at
/// the member boundary, not propagated — the cell is recorded failed with a
/// `Panic` kind, and the run record persists.
#[test]
fn execute_member_panic_is_contained_as_a_failed_cell() {
    let reg = temp_registry("member_panic");
    let doc = campaign(&["AAA", "BBB"]);
    let proc_doc = process(vec![sweep_stage(false)]);
    let runner = PanicRunner;
    let out = execute(CAMPAIGN_ID, &doc, &proc_doc, &strategies(), &runner, &reg, DEFAULT_PARALLEL_INSTRUMENTS)
        .expect("a member panic is contained, not a process abort");
    assert!(out.record.cells.iter().all(|c| c
        .fault
        .as_ref()
        .is_some_and(|f| f.kind == aura_registry::CellFaultKind::Panic)));
    assert_eq!(reg.load_campaign_runs().expect("load runs").len(), 1);
}

#[test]
fn execute_deflate_uses_campaign_seed() {
    let reg = temp_registry("deflate_seed");
    let doc = campaign(&["EURUSD"]); // seed: 7
    let proc_doc = process(vec![sweep_stage(true)]);
    let out = execute(CAMPAIGN_ID, &doc, &proc_doc, &strategies(), &FakeRunner::clean(), &reg, DEFAULT_PARALLEL_INSTRUMENTS)
        .expect("deflated sweep executes");
    let sel = out.record.cells[0].stages[0].selection.as_ref().expect("selection recorded");
    assert_eq!(sel.selection.mode, SelectionMode::Argmax);
    assert_eq!(sel.selection.seed, Some(7), "deflation nulls seed from the campaign doc");
    assert_eq!(sel.selection.n_resamples, Some(1000));
    assert_eq!(sel.selection.block_len, Some(5));
}

#[test]
fn execute_refuses_malformed_campaign_id() {
    let reg = temp_registry("bad_id");
    let doc = campaign(&["EURUSD"]);
    let proc_doc = process(vec![sweep_stage(false)]);
    let err = execute("short", &doc, &proc_doc, &strategies(), &FakeRunner::clean(), &reg, DEFAULT_PARALLEL_INSTRUMENTS)
        .expect_err("a non-64-hex campaign id is refused, not sliced");
    match err {
        ExecFault::PipelineShape { detail } => assert!(
            detail.contains("campaign_id must be a 64-hex content id"),
            "detail names the id contract: {detail}",
        ),
        other => panic!("expected ExecFault::PipelineShape, got {other:?}"),
    }
    assert!(reg.load_family_members().expect("load members").is_empty());
}

#[test]
fn execute_mc_after_gate_bootstraps_each_survivor() {
    let reg = temp_registry("mc_per_survivor");
    let doc = campaign(&["EURUSD"]); // seed: 7
    let proc_doc = process(vec![
        sweep_stage(false),
        gate_stage(0.3),
        StageBlock::MonteCarlo { resamples: 200, block_len: 2 },
    ]);
    let out = execute(CAMPAIGN_ID, &doc, &proc_doc, &strategies(), &FakeRunner::clean(), &reg, DEFAULT_PARALLEL_INSTRUMENTS)
        .expect("mc-after-gate campaign executes");

    // planted nets .26/.29/.36/.39 -> gate gt 0.3 keeps ordinals 2 and 3
    let cell = &out.record.cells[0];
    assert_eq!(cell.stages.len(), 3);
    let mc = &cell.stages[2];
    assert_eq!(mc.block, "std::monte_carlo");
    assert_eq!(mc.family_id, None);
    assert_eq!(mc.survivor_ordinals, None);
    assert_eq!(mc.selection, None);

    // each survivor's bootstrap == a hand-called r_bootstrap on its planted
    // net_trade_rs, seeded from the campaign doc (the deflation convention)
    let net = |fast: i64, slow: i64| (fast * 10 + slow) as f64 / 100.0;
    let expected = StageBootstrap::PerSurvivor(vec![
        (2, r_bootstrap(&planted_net_trade_rs(net(3, 6)), 200, 2, 7)),
        (3, r_bootstrap(&planted_net_trade_rs(net(3, 9)), 200, 2, 7)),
    ]);
    assert_eq!(mc.bootstrap, Some(expected));

    // the record round-trips through the store with the bootstrap intact
    let runs = reg.load_campaign_runs().expect("load campaign runs");
    assert_eq!(runs[0], out.record);
}

/// Property (#256 fork B): a `std::grid` first stage crosses the `run_cell`
/// seam as bare parameter points (`StageFlow::Points`) rather than executed
/// members — it persists no family of its own — and the immediately-following
/// `std::walk_forward` stage still sweeps exactly the full axis grid, picking
/// the same per-window winner a selection-carrying `std::sweep` feeding the
/// same wf stage would (`execute_mc_after_wf_pools_the_oos_series`'s
/// `sweep_stage()` + `wf_stage()` shape, minus the intermediate sweep family).
/// A regression that let `StageFlow::Points` drop or mis-order a point before
/// crossing the seam would flip the wf winner or the per-window report count
/// here even though the preflight placement tests (lib.rs) stay green.
#[test]
fn execute_grid_first_stage_feeds_walk_forward_with_all_grid_points() {
    let reg = temp_registry("grid_then_wf");
    let doc = campaign(&["EURUSD"]); // seed: 7, window [1_000, 9_000]
    let proc_doc = process(vec![StageBlock::Grid, wf_stage()]);
    let out = execute(CAMPAIGN_ID, &doc, &proc_doc, &strategies(), &FakeRunner::clean(), &reg, DEFAULT_PARALLEL_INSTRUMENTS)
        .expect("grid-then-wf campaign executes");

    // std::grid persists no family of its own — only std::walk_forward's does
    assert_eq!(out.cells[0].families.len(), 1);
    let wf = &out.cells[0].families[0];
    assert_eq!(wf.block, "std::walk_forward");

    // stage realization: std::grid first (no family/selection), then wf
    let cell = &out.record.cells[0];
    assert_eq!(cell.stages.len(), 2);
    assert_eq!(cell.stages[0].block, "std::grid");
    assert_eq!(cell.stages[0].family_id, None);
    assert_eq!(cell.stages[0].survivor_ordinals, None);
    assert_eq!(cell.stages[0].selection, None);
    assert_eq!(cell.stages[1].block, "std::walk_forward");

    // the roller yields 2 windows over the FULL 2x2 grid (nothing filtered
    // the population before wf); both pick the planted argmax (3,9)
    assert_eq!(wf.reports.len(), 2);
    let winner_params =
        vec![("fast".to_string(), Scalar::i64(3)), ("slow".to_string(), Scalar::i64(9))];
    for report in &wf.reports {
        assert_eq!(report.manifest.params, winner_params);
    }

    // the record round-trips through the store unchanged
    let runs = reg.load_campaign_runs().expect("load campaign runs");
    assert_eq!(runs[0], out.record);
}

#[test]
fn execute_mc_after_wf_pools_the_oos_series() {
    let reg = temp_registry("mc_pooled");
    let doc = campaign(&["EURUSD"]); // seed: 7, window [1_000, 9_000]
    let proc_doc = process(vec![
        sweep_stage(false),
        wf_stage(),
        StageBlock::MonteCarlo { resamples: 200, block_len: 3 },
    ]);
    let out = execute(CAMPAIGN_ID, &doc, &proc_doc, &strategies(), &FakeRunner::clean(), &reg, DEFAULT_PARALLEL_INSTRUMENTS)
        .expect("mc-after-wf campaign executes");

    // the roller yields 2 windows; both pick the planted argmax (3,9)
    let wf = &out.cells[0].families[1];
    assert_eq!(wf.block, "std::walk_forward");
    assert_eq!(wf.reports.len(), 2);

    // pooled input = the wf family reports' net_trade_rs concatenated in report
    // order (roll order per walkforward_member_reports)
    let mut pooled: Vec<f64> = Vec::new();
    for report in &wf.reports {
        pooled.extend_from_slice(&report.metrics.r.as_ref().expect("planted R block").net_trade_rs);
    }
    let net = (3 * 10 + 9) as f64 / 100.0;
    let mut expected_pool = planted_net_trade_rs(net);
    expected_pool.extend(planted_net_trade_rs(net));
    assert_eq!(pooled, expected_pool);

    let mc = &out.record.cells[0].stages[2];
    assert_eq!(mc.block, "std::monte_carlo");
    assert_eq!(mc.family_id, None);
    assert_eq!(mc.bootstrap, Some(StageBootstrap::PooledOos(r_bootstrap(&pooled, 200, 3, 7))));
}

#[test]
fn execute_mc_zero_trade_member_records_degenerate() {
    let reg = temp_registry("mc_zero_trade");
    let doc = campaign(&["EURUSD"]); // seed: 7
    let proc_doc =
        process(vec![sweep_stage(false), StageBlock::MonteCarlo { resamples: 200, block_len: 2 }]);
    // ordinal 0 == odometer point (2,6): its report carries no R block at all
    let runner = ZeroTradeRunner {
        inner: FakeRunner::clean(),
        strip: vec![("fast".to_string(), Scalar::i64(2)), ("slow".to_string(), Scalar::i64(6))],
    };
    let out = execute(CAMPAIGN_ID, &doc, &proc_doc, &strategies(), &runner, &reg, DEFAULT_PARALLEL_INSTRUMENTS)
        .expect("a zero-trade member is a valid result, not a fault");

    let mc = &out.record.cells[0].stages[1];
    assert_eq!(mc.block, "std::monte_carlo");
    let Some(StageBootstrap::PerSurvivor(entries)) = &mc.bootstrap else {
        panic!("expected a per-survivor bootstrap, got {:?}", mc.bootstrap);
    };
    // no gate ran: all 4 members survive, each annotated
    assert_eq!(entries.iter().map(|(o, _)| *o).collect::<Vec<_>>(), vec![0, 1, 2, 3]);
    // the r: None member records the engine's defined all-zero degenerate
    assert_eq!(entries[0].1, r_bootstrap(&[], 200, 2, 7));
    assert_eq!(entries[0].1.n_trades, 0);
    assert_eq!(entries[0].1.e_r.mean, 0.0);
    assert_eq!(entries[0].1.prob_le_zero, 0.0);
}

#[test]
fn execute_generalize_across_two_instruments() {
    let reg = temp_registry("generalize_two");
    // the fake negates "AAA" nets -> divergent winners across instruments
    let doc = campaign(&["AAA", "BBB"]); // seed: 7
    let proc_doc = process(vec![sweep_stage(false), wf_stage(), generalize_stage()]);
    let out = execute(CAMPAIGN_ID, &doc, &proc_doc, &strategies(), &FakeRunner::clean(), &reg, DEFAULT_PARALLEL_INSTRUMENTS)
        .expect("generalize campaign executes");

    assert_eq!(out.record.generalizations.len(), 1);
    let g = &out.record.generalizations[0];
    assert_eq!((g.strategy_ordinal, g.window_ordinal), (0, 0));
    assert!(g.missing.is_empty());

    // winners carry each instrument's nominated params (the last wf window's
    // chosen point): AAA argmaxes its NEGATED nets -> (2,6); BBB -> (3,9)
    assert_eq!(
        g.winners,
        vec![
            (
                "AAA".to_string(),
                vec![("fast".to_string(), Scalar::i64(2)), ("slow".to_string(), Scalar::i64(6))],
            ),
            (
                "BBB".to_string(),
                vec![("fast".to_string(), Scalar::i64(3)), ("slow".to_string(), Scalar::i64(9))],
            ),
        ],
    );

    // the stored grade == the shipped generalization() hand-called over the
    // two nominated reports (cells are in instrument order: AAA then BBB)
    let pairs: Vec<(String, &RunReport)> = out
        .cells
        .iter()
        .zip(["AAA", "BBB"])
        .map(|(cell, inst)| {
            let (_, report) = cell.nominee.as_ref().expect("nominee present");
            (inst.to_string(), report)
        })
        .collect();
    let expected = generalization(&pairs, "net_expectancy_r").expect("hand-called generalization");
    assert_eq!(g.generalization.as_ref(), Some(&expected));

    // divergence is exposed, never averaged: worst case is AAA's negated net
    let aaa_net = -((2 * 10 + 6) as f64 / 100.0);
    let bbb_net = (3 * 10 + 9) as f64 / 100.0;
    assert_eq!(
        expected.per_instrument,
        vec![("AAA".to_string(), aaa_net), ("BBB".to_string(), bbb_net)],
    );
    assert_eq!(expected.worst_case, aaa_net);
    assert_eq!(expected.sign_agreement, 1);

    // the record round-trips through the store with the generalization intact
    let runs = reg.load_campaign_runs().expect("load campaign runs");
    assert_eq!(runs[0], out.record);
}

#[test]
fn execute_generalize_shortfall_records_missing() {
    let reg = temp_registry("generalize_shortfall");
    // AAA's planted nets are all negative -> the gt-0 gate empties its cell
    let doc = campaign(&["AAA", "BBB"]);
    let proc_doc = process(vec![sweep_stage(false), gate_stage(0.0), generalize_stage()]);
    let out = execute(CAMPAIGN_ID, &doc, &proc_doc, &strategies(), &FakeRunner::clean(), &reg, DEFAULT_PARALLEL_INSTRUMENTS)
        .expect("a truncated cell is a recorded shortfall, not a fault");

    assert!(out.cells[0].nominee.is_none(), "gate-truncated cell nominates nothing");
    assert_eq!(out.record.generalizations.len(), 1);
    let g = &out.record.generalizations[0];
    assert_eq!(g.generalization, None, "fewer than 2 winners -> no grade computed");
    assert_eq!(g.missing, vec!["AAA".to_string()]);
    assert_eq!(g.winners.len(), 1);
    assert_eq!(g.winners[0].0, "BBB");
    assert_eq!(
        g.winners[0].1,
        vec![("fast".to_string(), Scalar::i64(3)), ("slow".to_string(), Scalar::i64(9))],
    );
}

#[test]
fn execute_generalize_only_after_sweep() {
    let reg = temp_registry("generalize_sweep_only");
    let doc = campaign(&["EURUSD", "GER40"]);
    let proc_doc = process(vec![sweep_stage(false), generalize_stage()]);
    let out = execute(CAMPAIGN_ID, &doc, &proc_doc, &strategies(), &FakeRunner::clean(), &reg, DEFAULT_PARALLEL_INSTRUMENTS)
        .expect("sweep-only generalize campaign executes");

    // no wf ran: each cell nominates its sweep winner (planted argmax (3,9))
    let winner_params =
        vec![("fast".to_string(), Scalar::i64(3)), ("slow".to_string(), Scalar::i64(9))];
    for cell in &out.cells {
        let (params, report) = cell.nominee.as_ref().expect("sweep winner nominated");
        assert_eq!(params, &winner_params);
        assert_eq!(report.manifest.params, winner_params);
    }

    let g = &out.record.generalizations[0];
    assert_eq!(
        g.winners,
        vec![
            ("EURUSD".to_string(), winner_params.clone()),
            ("GER40".to_string(), winner_params.clone()),
        ],
    );
    assert!(g.missing.is_empty());
    let grade = g.generalization.as_ref().expect("two winners -> graded");
    let net = (3 * 10 + 9) as f64 / 100.0;
    assert_eq!(grade.n_instruments, 2);
    assert_eq!(grade.sign_agreement, 2);
    assert_eq!(grade.worst_case, net);
}

/// Property (0108 acceptance criterion 1, the spec's worked example shape
/// `sweep, gate, walk_forward, monte_carlo, generalize`): the two terminal
/// annotators compose in ONE pipeline without disturbing each other or the
/// population stages' generalize-nominee. `std::monte_carlo` sits between
/// `std::walk_forward` and `std::generalize` in the pipeline, yet its
/// per-cell bootstrap still pools exactly the wf family's OOS `net_trade_rs`
/// (unaffected by generalize running after it), and `std::generalize`'s
/// campaign-scope nominee is still the wf stage's last-window winner
/// (unaffected by mc having annotated the cell first). A regression that
/// let one annotator's bookkeeping leak into or clobber the other's input
/// (e.g. mc consuming the nominee, or generalize reading mc's output)
/// would flip an assertion here even though each annotator's own isolated
/// test (above) stays green.
#[test]
fn execute_mc_and_generalize_compose_in_one_pipeline() {
    let reg = temp_registry("mc_and_generalize_compose");
    let doc = campaign(&["AAA", "BBB"]); // seed: 7
    let proc_doc = process(vec![
        sweep_stage(false),
        gate_stage(-1.0), // gt -1.0: every planted net (AAA negated too) survives
        wf_stage(),
        StageBlock::MonteCarlo { resamples: 200, block_len: 2 },
        generalize_stage(),
    ]);
    let out = execute(CAMPAIGN_ID, &doc, &proc_doc, &strategies(), &FakeRunner::clean(), &reg, DEFAULT_PARALLEL_INSTRUMENTS)
        .expect("full v2 pipeline executes");

    // both cells realize all four per-cell stages; generalize contributes no
    // per-cell stage (it is the campaign-scope phase after the cell loop)
    for cell in &out.record.cells {
        assert_eq!(cell.stages.len(), 4);
        assert_eq!(cell.stages[3].block, "std::monte_carlo");
    }

    // mc's bootstrap pools the wf family's OOS net_trade_rs exactly as it does
    // with no generalize stage after it (execute_mc_after_wf_pools_the_oos_series)
    let wf = &out.cells[0].families[1];
    assert_eq!(wf.block, "std::walk_forward");
    let mut pooled: Vec<f64> = Vec::new();
    for report in &wf.reports {
        pooled.extend_from_slice(&report.metrics.r.as_ref().expect("planted R block").net_trade_rs);
    }
    let expected_mc = StageBootstrap::PooledOos(r_bootstrap(&pooled, 200, 2, 7));
    assert_eq!(out.record.cells[0].stages[3].bootstrap, Some(expected_mc));

    // generalize still nominates the wf stage's last-window winner per
    // instrument, sourced independently of the mc stage between them
    assert_eq!(out.record.generalizations.len(), 1);
    let g = &out.record.generalizations[0];
    assert!(g.missing.is_empty(), "the -1.0 gate keeps every member on both instruments");
    // AAA's nets are negated by the fake -> argmax picks the LEAST-negative
    // point (2,6); BBB argmaxes its unmodified nets at (3,9) (same as
    // execute_generalize_across_two_instruments).
    assert_eq!(
        g.winners[0],
        ("AAA".to_string(), vec![("fast".to_string(), Scalar::i64(2)), ("slow".to_string(), Scalar::i64(6))]),
    );
    assert_eq!(
        g.winners[1],
        ("BBB".to_string(), vec![("fast".to_string(), Scalar::i64(3)), ("slow".to_string(), Scalar::i64(9))]),
    );
    assert!(g.generalization.is_some(), "two instruments -> graded");

    // the record round-trips through the store with BOTH annotations intact
    let runs = reg.load_campaign_runs().expect("load campaign runs");
    assert_eq!(runs[0], out.record);
}

/// The 0109 name contract (spec seam note / #201 d5): `execute` claims a
/// TraceStore family name iff the document requests persist_taps — the
/// RETURNED record and the STORED line both carry the same derived
/// `Some("{campaign8}-{run}")` (`append_campaign_run` composes it onto the
/// stored line from the claim sentinel; `execute` mirrors the same
/// derivation onto the returned copy after the counter is assigned). The
/// empty-taps `None` side is pinned on
/// `execute_sweep_only_records_family_and_selection`.
#[test]
fn execute_persist_taps_stamps_trace_name() {
    let reg = temp_registry("persist_taps_trace_name");
    let mut doc = campaign(&["EURUSD"]);
    doc.presentation.persist_taps = vec!["equity".to_string()];
    let proc_doc = process(vec![sweep_stage(false)]);
    let out = execute(CAMPAIGN_ID, &doc, &proc_doc, &strategies(), &FakeRunner::clean(), &reg, DEFAULT_PARALLEL_INSTRUMENTS)
        .expect("persist_taps campaign executes");

    // returned record: the derived name over the first run counter
    let expected = format!("{}-0", &CAMPAIGN_ID[..8]);
    assert_eq!(
        out.record.trace_name.as_deref(),
        Some(expected.as_str()),
        "the returned record carries the derived trace name",
    );

    // the stored line agrees — in the field and as a whole record
    let runs = reg.load_campaign_runs().expect("load campaign runs");
    assert_eq!(runs.len(), 1);
    assert_eq!(runs[0].trace_name.as_deref(), Some(expected.as_str()));
    assert_eq!(runs[0], out.record, "stored line and returned record are the same record");

    // a second run of the same campaign derives the next counter's name
    let second = execute(CAMPAIGN_ID, &doc, &proc_doc, &strategies(), &FakeRunner::clean(), &reg, DEFAULT_PARALLEL_INSTRUMENTS)
        .expect("second persist_taps run");
    let expected_1 = format!("{}-1", &CAMPAIGN_ID[..8]);
    assert_eq!(second.record.trace_name.as_deref(), Some(expected_1.as_str()));
}

/// Property (#210 T3): a campaign document's `risk` list is a structural
/// matrix axis, expanded like instruments/windows — each regime runs its own
/// full per-cell pipeline (its own cell, its own sweep family), distinguished
/// by a `-r{regime_ordinal}` family-name segment.
#[test]
fn execute_runs_one_family_per_regime_with_distinct_ids() {
    let reg = temp_registry("two_regimes");
    let mut doc = campaign(&["EURUSD"]);
    doc.risk = vec![
        aura_research::RiskRegime::Vol { length: 3, k: 1.5 },
        aura_research::RiskRegime::Vol { length: 3, k: 2.0 },
    ];
    let proc_doc = process(vec![sweep_stage(false)]);
    let out = execute(CAMPAIGN_ID, &doc, &proc_doc, &strategies(), &FakeRunner::clean(), &reg, DEFAULT_PARALLEL_INSTRUMENTS)
        .expect("two-regime campaign executes");

    // one cell per regime, each with its own sweep family.
    assert_eq!(out.cells.len(), 2, "one cell per regime");
    let p = &CAMPAIGN_ID[..8];
    assert_eq!(out.cells[0].families[0].family_id, format!("{p}-0-EURUSD-w0-r0-s0-0"));
    assert_eq!(out.cells[1].families[0].family_id, format!("{p}-0-EURUSD-w0-r1-s0-0"));
}

/// Property (#210 T3, C10 enforcement): generalize is graded PER REGIME, never
/// collapsed or cross-regime argmax'd — a regression that dropped
/// `regime_ordinal` from the campaign-scope nominee key (the `nominees`
/// `BTreeMap` in `execute`) would merge both regimes' per-instrument nominees
/// into ONE generalization group instead of two, doubling `winners` (with a
/// duplicate instrument name per group) rather than keeping each regime's
/// grade separate. Two instruments x two regimes, no wf stage (sweep winner
/// nominates directly) isolates the keying from any other stage's behaviour.
#[test]
fn execute_generalize_keeps_regimes_separate() {
    let reg = temp_registry("generalize_regimes");
    let mut doc = campaign(&["AAA", "BBB"]);
    doc.risk = vec![
        aura_research::RiskRegime::Vol { length: 3, k: 1.5 },
        aura_research::RiskRegime::Vol { length: 3, k: 2.0 },
    ];
    let proc_doc = process(vec![sweep_stage(false), generalize_stage()]);
    let out = execute(CAMPAIGN_ID, &doc, &proc_doc, &strategies(), &FakeRunner::clean(), &reg, DEFAULT_PARALLEL_INSTRUMENTS)
        .expect("two-regime generalize campaign executes");

    // 2 instruments x 2 regimes -> 4 cells, but exactly 2 generalize groups
    // (one per regime), each grading exactly the 2 instruments in ITS regime.
    assert_eq!(out.cells.len(), 4);
    assert_eq!(out.record.generalizations.len(), 2, "one generalization per regime");
    let regime_ordinals: Vec<usize> =
        out.record.generalizations.iter().map(|g| g.regime_ordinal).collect();
    assert_eq!(regime_ordinals, vec![0, 1], "each regime keeps its own generalize group");
    for g in &out.record.generalizations {
        assert_eq!(g.winners.len(), 2, "each regime's group grades both instruments, not both regimes' nominees combined");
        let instruments: Vec<&str> = g.winners.iter().map(|(inst, _)| inst.as_str()).collect();
        assert_eq!(instruments, vec!["AAA", "BBB"], "no cross-regime duplicate instrument entries");
        assert!(g.missing.is_empty());
        assert!(g.generalization.is_some(), "two instruments -> graded");
    }
}

/// C1 across the parallel cell loop: the same campaign under a 1-thread pool
/// with K=1 and an 8-thread pool with K=2 (fresh registry each) yields an
/// identical campaign-run record — the campaign-path sibling of the engine's
/// `*_is_deterministic_across_thread_counts` tests.
#[test]
fn campaign_outcome_is_deterministic_across_thread_counts_and_bounds() {
    let doc = campaign(&["AAA", "BBB", "CCC"]);
    let proc_doc = process(vec![sweep_stage(false)]);
    let run_with = |threads: usize, k: usize, name: &str| {
        let reg = temp_registry(name);
        let pool = rayon::ThreadPoolBuilder::new()
            .num_threads(threads)
            .build()
            .expect("build pool");
        pool.install(|| {
            execute(
                CAMPAIGN_ID,
                &doc,
                &proc_doc,
                &strategies(),
                &FakeRunner::clean(),
                &reg,
                std::num::NonZeroUsize::new(k).expect("nonzero"),
            )
        })
        .expect("campaign run")
    };
    let serial = run_with(1, 1, "det_across_serial");
    let parallel = run_with(8, 2, "det_across_parallel");
    assert_eq!(serial.run, parallel.run);
    assert_eq!(serial.record, parallel.record);
}

/// #277: a non-containable fault (here, `append_family`'s registry-level IO
/// refusal, never routed through `contain` since it is neither a
/// `MemberFault` nor a `WindowFault`) still propagates as a global `Err` out
/// of the parallel abort-latch loop, and does so with IDENTICAL content
/// regardless of thread count or pool size. Every cell's sweep hits the SAME
/// blocked family-store path, so whichever completed cell's `Err` the
/// doc-order scan in `execute` happens to pick (scheduling-dependent per the
/// exec.rs doc comment) carries the same IO error kind — exercising the
/// abort-latch + lowest-doc-order-fault-scan path (and never the
/// `unreachable!` rebuild arm, since this run returns `Err` before reaching
/// the rebuild) that the fault-free determinism test above does not.
#[test]
fn execute_propagates_a_non_containable_fault_deterministically_under_parallel_chunks() {
    let doc = campaign(&["AAA", "BBB", "CCC"]);
    let proc_doc = process(vec![sweep_stage(false)]);
    let run_with = |threads: usize, k: usize, name: &str| {
        let reg = temp_registry_with_blocked_family_store(name);
        let pool = rayon::ThreadPoolBuilder::new()
            .num_threads(threads)
            .build()
            .expect("build pool");
        pool.install(|| {
            execute(
                CAMPAIGN_ID,
                &doc,
                &proc_doc,
                &strategies(),
                &FakeRunner::clean(),
                &reg,
                std::num::NonZeroUsize::new(k).expect("nonzero"),
            )
        })
    };
    let serial = run_with(1, 1, "fault_prop_serial")
        .expect_err("a blocked family store is refused, not silently written");
    let parallel = run_with(8, 3, "fault_prop_parallel")
        .expect_err("same refusal under real cross-cell parallelism");
    let kind = |e: &ExecFault| match e {
        ExecFault::Registry(RegistryError::Io(io)) => io.kind(),
        other => panic!("expected ExecFault::Registry(Io), got {other:?}"),
    };
    assert_eq!(kind(&serial), std::io::ErrorKind::IsADirectory);
    assert_eq!(kind(&serial), kind(&parallel), "same fault kind regardless of scheduling");
}

/// Wraps the clean fake and tracks how many DISTINCT instruments are inside
/// `run_member` at once (refcounted per instrument; high-water mark in
/// `max_seen`). The 5 ms hold makes overlap near-certain on an 8-thread pool
/// if the residency bound is absent.
struct CountingRunner {
    inner: FakeRunner,
    live: std::sync::Mutex<BTreeMap<String, usize>>,
    max_seen: std::sync::atomic::AtomicUsize,
}

impl MemberRunner for CountingRunner {
    fn run_member(
        &self,
        cell: &CellSpec,
        params: &[(String, Scalar)],
        window_ms: (i64, i64),
    ) -> Result<RunReport, MemberFault> {
        let distinct = {
            let mut live = self.live.lock().expect("live instrument map");
            *live.entry(cell.instrument.clone()).or_insert(0) += 1;
            live.len()
        };
        self.max_seen.fetch_max(distinct, std::sync::atomic::Ordering::SeqCst);
        std::thread::sleep(std::time::Duration::from_millis(5));
        let out = self.inner.run_member(cell, params, window_ms);
        let mut live = self.live.lock().expect("live instrument map");
        let count = live.get_mut(&cell.instrument).expect("entered above");
        *count -= 1;
        if *count == 0 {
            live.remove(&cell.instrument);
        }
        out
    }
}

/// The structural residency bound: with K=1 on an 8-thread pool, no two
/// distinct instruments are ever inside `run_member` at the same time.
#[test]
fn parallel_cells_never_exceed_the_instrument_bound() {
    let reg = temp_registry("instrument_bound");
    let doc = campaign(&["AAA", "BBB", "CCC"]);
    let proc_doc = process(vec![sweep_stage(false)]);
    let runner = CountingRunner {
        inner: FakeRunner::clean(),
        live: std::sync::Mutex::new(BTreeMap::new()),
        max_seen: std::sync::atomic::AtomicUsize::new(0),
    };
    let pool = rayon::ThreadPoolBuilder::new().num_threads(8).build().expect("build pool");
    pool.install(|| {
        execute(
            CAMPAIGN_ID,
            &doc,
            &proc_doc,
            &strategies(),
            &runner,
            &reg,
            std::num::NonZeroUsize::new(1).expect("nonzero"),
        )
    })
    .expect("campaign run");
    let max = runner.max_seen.load(std::sync::atomic::Ordering::SeqCst);
    assert!(max <= 1, "K=1 must bound distinct live instruments to 1, saw {max}");
}

/// A registry-write fault is infrastructure, not a cell pathology: it stays
/// run-fatal in the parallel loop (no cell containment, no partial
/// campaign-run record).
#[test]
fn a_registry_fault_is_run_fatal_in_the_parallel_loop() {
    let dir = std::path::Path::new(env!("CARGO_TARGET_TMPDIR"))
        .join("aura-campaign-exec-registry_fatal");
    let _ = std::fs::remove_dir_all(&dir);
    std::fs::create_dir_all(&dir).expect("create temp registry dir");
    // A directory squatting on the family store makes every append_family fail.
    std::fs::create_dir(dir.join("families.jsonl")).expect("squat the family store");
    let reg = Registry::open(dir.join("runs.jsonl"));
    let doc = campaign(&["AAA", "BBB"]);
    let proc_doc = process(vec![sweep_stage(false)]);
    let err = execute(
        CAMPAIGN_ID,
        &doc,
        &proc_doc,
        &strategies(),
        &FakeRunner::clean(),
        &reg,
        DEFAULT_PARALLEL_INSTRUMENTS,
    )
    .expect_err("a dead family store is run-fatal");
    assert!(matches!(err, ExecFault::Registry(_)));
    assert!(
        !dir.join("campaign_runs.jsonl").exists(),
        "no partial campaign-run record on the fatal path"
    );
}

/// #272 under the parallel loop: member faults stay contained as failed
/// cells across concurrently-running cells — the run record persists with
/// every cell recorded, never a global abort.
#[test]
fn member_faults_stay_contained_in_the_parallel_loop() {
    let reg = temp_registry("parallel_containment");
    let doc = campaign(&["AAA", "BBB", "CCC"]);
    let proc_doc = process(vec![sweep_stage(false)]);
    let point = |fast: i64, slow: i64| {
        vec![("fast".to_string(), Scalar::i64(fast)), ("slow".to_string(), Scalar::i64(slow))]
    };
    let runner = FakeRunner {
        faults: vec![(point(2, 9), MemberFault::Run("boom".to_string()))],
    };
    let pool = rayon::ThreadPoolBuilder::new().num_threads(8).build().expect("build pool");
    let out = pool
        .install(|| {
            execute(
                CAMPAIGN_ID,
                &doc,
                &proc_doc,
                &strategies(),
                &runner,
                &reg,
                std::num::NonZeroUsize::new(2).expect("nonzero"),
            )
        })
        .expect("member faults are contained, not a global abort");
    assert_eq!(out.record.cells.len(), 3, "every cell is recorded");
    assert!(out.record.cells.iter().all(|c| c.fault.is_some()));
    assert_eq!(reg.load_campaign_runs().expect("load runs").len(), 1);
}

/// Defense in depth (audit tidy, #277): `execute` itself refuses duplicate
/// instruments via `preflight`, independent of the CLI's validate tier — a
/// direct caller skipping `validate_campaign` must not reach the cell loop
/// with colliding registry family names.
#[test]
fn execute_refuses_duplicate_instruments() {
    let reg = temp_registry("duplicate_instruments");
    let doc = campaign(&["AAA", "AAA"]);
    let proc_doc = process(vec![sweep_stage(false)]);
    let err = execute(
        CAMPAIGN_ID,
        &doc,
        &proc_doc,
        &strategies(),
        &FakeRunner::clean(),
        &reg,
        DEFAULT_PARALLEL_INSTRUMENTS,
    )
    .expect_err("duplicate instruments collide on family names");
    assert!(matches!(err, ExecFault::PipelineShape { .. }));
}