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};
use aura_core::{Scalar, ScalarKind, Timestamp};
use aura_engine::{RMetrics, RunManifest, RunMetrics, RunReport, SelectionMode};
use aura_registry::{FamilyKind, Registry};
use aura_research::{
    Axis, CampaignDoc, Cmp, DataSection, DocKind, DocRef, Predicate, Presentation, ProcessDoc,
    ProcessRef, SelectRule, StageBlock, StrategyEntry, 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")
}

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(),
            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],
                trade_rs: Vec::new(),
            }),
        },
    }
}

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

/// 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 }],
        },
        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 {
        metric: "net_expectancy_r".to_string(),
        select: SelectRule::Argmax,
        deflate,
    }
}

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

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::env::temp_dir()
        .join(format!("aura-campaign-exec-{}-{}", std::process::id(), 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)
        .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-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].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");
}

#[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)
        .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)
        .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]));
}

#[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)
        .expect("first run");
    let second = execute(CAMPAIGN_ID, &doc, &proc_doc, &strategies(), &FakeRunner::clean(), &reg)
        .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-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);
}

#[test]
fn execute_member_fault_aborts_with_lowest_index() {
    let reg = temp_registry("member_fault");
    let doc = campaign(&["EURUSD"]);
    let proc_doc = process(vec![sweep_stage(false)]);
    // faults planted at enumeration indices 1 (2,9) and 2 (3,6)
    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 at index 1".to_string())),
            (point(3, 6), MemberFault::Run("boom at index 2".to_string())),
        ],
    };
    let err = execute(CAMPAIGN_ID, &doc, &proc_doc, &strategies(), &runner, &reg)
        .expect_err("a faulted member aborts the run");
    match err {
        ExecFault::Member(fault) => assert_eq!(
            fault,
            MemberFault::Run("boom at index 1".to_string()),
            "the LOWEST enumeration index's fault is reported",
        ),
        other => panic!("expected ExecFault::Member, got {other:?}"),
    }
    // the fault aborted before any family or realization write
    assert!(reg.load_family_members().expect("load members").is_empty());
    assert!(reg.load_campaign_runs().expect("load campaign runs").is_empty());
}

#[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)
        .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)
        .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());
}