diff --git a/crates/aura-campaign/src/exec.rs b/crates/aura-campaign/src/exec.rs new file mode 100644 index 0000000..7cea624 --- /dev/null +++ b/crates/aura-campaign/src/exec.rs @@ -0,0 +1,603 @@ +//! 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::{ + sweep, walk_forward, GridSpace, ListSpace, RollMode, RunManifest, RunMetrics, RunReport, + Space, SweepFamily, SweepPoint, WindowBounds, WindowRoller, WindowRun, +}; +use aura_registry::{ + optimize, optimize_deflated, optimize_plateau, sweep_member_reports, + walkforward_member_reports, CampaignRunRecord, CellRealization, FamilyKind, PlateauMode, + Registry, 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, +} + +/// 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, + pub selections: Vec, +} + +/// 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, +} + +/// 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 { + 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| b.is_ascii_hexdigit()) { + 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 = Vec::new(); + let mut cells_rec: Vec = Vec::new(); + 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() { + 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), + }; + let (outcome, realization) = run_cell( + &cell, + process, + campaign_prefix, + window_ordinal, + campaign.seed, + runner, + registry, + )?; + cells_out.push(outcome); + cells_rec.push(realization); + } + } + } + + let mut record = CampaignRunRecord { + campaign: campaign_id.to_string(), + process: process_id, + run: 0, + seed: campaign.seed, + cells: cells_rec, + }; + let run = registry.append_campaign_run(&record).map_err(ExecFault::Registry)?; + record.run = run; + 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 = Vec::new(); + let mut selections: Vec = Vec::new(); + let mut stages: Vec = 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, RunReport)> = Vec::new(); + + for (stage_ordinal, stage) in process.pipeline.iter().enumerate() { + match stage { + StageBlock::Sweep { metric, select, deflate } => { + // Deterministic, self-describing family name (the "-{run}" + // suffix is appended by the registry). + let family_name = format!( + "{campaign_prefix}-{}-{}-w{window_ordinal}-s{stage_ordinal}", + cell.strategy_ordinal, cell.instrument, + ); + 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)?; + let (winner, selection) = + select_sweep_winner(&family, &grid.axis_lens, metric, *select, *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(), + }); + stages.push(StageRealization { + block: "std::sweep".to_string(), + family_id: Some(family_id.clone()), + survivor_ordinals: None, + selection: Some(StageSelection { winner_ordinal, params, selection }), + }); + // 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 = 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, + }); + if empty { + 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}-s{stage_ordinal}", + cell.strategy_ordinal, cell.instrument, + ); + // 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> = + survivors.iter().map(|(_, point, _)| point.clone()).collect(); + let fam = run_walk_forward_stage( + seed, + cell, + stage_ordinal, + stage, + &grid.specs, + &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, + }); + families.push(fam); + } + StageBlock::MonteCarlo { .. } | StageBlock::Generalize { .. } => { + unreachable!("preflight refuses non-v1 stages before any member runs") + } + } + } + + Ok(( + CellOutcome { families, selections }, + CellRealization { + strategy: cell.strategy_id.clone(), + instrument: cell.instrument.clone(), + window_ms: cell.window_ms, + 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, + axis_lens: Vec, + points: Vec>, +} + +/// 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) -> SweepGrid { + let specs: Vec = axes + .iter() + .map(|(name, axis)| ParamSpec { name: name.clone(), kind: axis.kind }) + .collect(); + let values: Vec> = 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> = 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 { + 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> = + grid.points.iter().map(|p| p.iter().map(|s| s.cell()).collect()).collect(); + let faults: Mutex> = Mutex::new(Vec::new()); + let family = sweep(&space, |point: &[Cell]| { + let params = zip_params(&grid.specs, point); + match runner.run_member(cell, ¶ms, 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(¶ms, 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(), + 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 sweep stage's winner + selection provenance. 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`). +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 { + space.iter().zip(point).map(|(ps, c)| Scalar::from_cell(ps.kind, *c)).collect() +} + +/// 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, pick +/// the winner (argmax with trials-deflation provenance, seeded from the +/// campaign seed — the shipped select_winner convention; plateau in +/// walk_forward is preflight-refused), 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], + survivors: &[Vec], + family_name: &str, + runner: &dyn MemberRunner, + registry: &Registry, +) -> Result { + let StageBlock::WalkForward { in_sample_ms, out_of_sample_ms, step_ms, mode, metric, .. } = + 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, detail: format!("{e:?}") })?; + // 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 = 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> = 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> = 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: argmax with trials-deflation provenance, seeded from the + // campaign seed (the shipped select_winner convention). + let (winner, selection) = match optimize_deflated( + &is_family, + metric, + DEFLATION_N_RESAMPLES, + DEFLATION_BLOCK_LEN, + seed, + ) { + Ok(picked) => picked, + Err(e) => { + faults.lock().unwrap().push((widx, ExecFault::Registry(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 }) +} diff --git a/crates/aura-campaign/src/lib.rs b/crates/aura-campaign/src/lib.rs index 3cd2d16..050ae9d 100644 --- a/crates/aura-campaign/src/lib.rs +++ b/crates/aura-campaign/src/lib.rs @@ -11,6 +11,9 @@ //! realizes one campaign document; it owns no topology, no data sources, //! and no UI. +mod exec; +pub use exec::{execute, CampaignOutcome, CellOutcome, StageFamily, StageSelectionOut}; + use std::collections::BTreeMap; use aura_core::Scalar; @@ -143,8 +146,6 @@ pub fn member_metric(report: &RunReport, name: &str) -> Option { } /// Whether `value threshold` holds — the gate's comparator arm. -// Consumed by `execute`'s gate stage; the allow drops once that lands. -#[allow(dead_code)] fn predicate_holds(cmp: &Cmp, value: f64, threshold: f64) -> bool { match cmp { Cmp::Gt => value > threshold, @@ -163,8 +164,6 @@ fn predicate_holds(cmp: &Cmp, value: f64, threshold: f64) -> bool { /// composes only with `argmax`; walk_forward lengths must fit `i64` (the /// roller's Timestamp unit). The campaign parameter is the seam for /// campaign-level static checks (none in v1, hence unused). -// Consumed by `execute`; the allow drops once that lands. -#[allow(dead_code)] pub(crate) fn preflight(process: &ProcessDoc, _campaign: &CampaignDoc) -> Result<(), ExecFault> { // v1 boundary first: an mc/generalize stage refuses wherever it sits, // before any shape complaint about the same stage. @@ -583,3 +582,445 @@ mod tests { } } } + +#[cfg(test)] +mod wf_tests { + use std::collections::BTreeMap; + use std::sync::Mutex; + + use aura_analysis::SelectionMode; + use aura_core::{Scalar, ScalarKind}; + use aura_engine::{RollMode, RunManifest, RunMetrics, RunReport, Timestamp, WindowRoller}; + use aura_registry::{FamilyKind, Registry}; + use aura_research::{ + Axis, CampaignDoc, Cmp, DataSection, DocKind, DocRef, Predicate, Presentation, + ProcessDoc, ProcessRef, SelectRule, StageBlock, StrategyEntry, WfMode, Window, + }; + + use super::{execute, CampaignOutcome, CellSpec, ExecFault, MemberFault, MemberRunner}; + + /// A per-test registry in a fresh temp dir (a Registry's family store + /// isolates per DIRECTORY, not per filename — see Registry::open). + fn wf_registry(name: &str) -> Registry { + let dir = std::env::temp_dir() + .join(format!("aura-campaign-wf-{}-{}", std::process::id(), name)); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).expect("temp dir"); + Registry::open(dir.join("runs.jsonl")) + } + + /// One strategy x one instrument x one window, over a single I64 axis + /// "len" with values [1, 2, 3, 4] (the fake runner scores total_pips == + /// the len value, so ranking and gating are fully determined). + fn wf_campaign(window: (i64, i64)) -> CampaignDoc { + CampaignDoc { + format_version: 1, + kind: DocKind::Campaign, + name: "wf-test".to_string(), + description: None, + data: DataSection { + instruments: vec!["SYNTH".to_string()], + windows: vec![Window { from_ms: window.0, to_ms: window.1 }], + }, + strategies: vec![StrategyEntry { + r#ref: DocRef::ContentId("c".repeat(64)), + axes: BTreeMap::from([( + "len".to_string(), + Axis { + kind: ScalarKind::I64, + values: vec![ + Scalar::i64(1), + Scalar::i64(2), + Scalar::i64(3), + Scalar::i64(4), + ], + }, + )]), + }], + process: ProcessRef { r#ref: DocRef::ContentId("d".repeat(64)) }, + seed: 7, + presentation: Presentation { persist_taps: vec![], emit: vec![] }, + } + } + + /// sweep -> (optional total_pips-gt gate) -> walk_forward, both ranked + /// stages on total_pips / argmax, rolling mode. + fn wf_process(gate_gt: Option, is_ms: u64, oos_ms: u64, step_ms: u64) -> ProcessDoc { + let mut pipeline = vec![StageBlock::Sweep { + metric: "total_pips".to_string(), + select: SelectRule::Argmax, + deflate: false, + }]; + if let Some(threshold) = gate_gt { + pipeline.push(StageBlock::Gate { + all: vec![Predicate { + metric: "total_pips".to_string(), + cmp: Cmp::Gt, + value: threshold, + }], + }); + } + pipeline.push(StageBlock::WalkForward { + in_sample_ms: is_ms, + out_of_sample_ms: oos_ms, + step_ms, + mode: WfMode::Rolling, + metric: "total_pips".to_string(), + select: SelectRule::Argmax, + }); + ProcessDoc { + format_version: 1, + kind: DocKind::Process, + name: "wf-proc".to_string(), + description: None, + pipeline, + } + } + + /// One (window bounds, params) -> fault mapping; a type alias rather than + /// a struct-level `#[allow]` since `WfFakeRunner::faulty`'s own parameter + /// re-triggers the type-complexity lint independently of the field. + type WfFault = ((i64, i64), Vec<(String, Scalar)>, MemberFault); + + /// Deterministic fake member runner keyed on (window bounds, params): + /// total_pips == the summed numeric param values (here: the single "len" + /// value), the report's manifest.window echoes the window it was run over, + /// and every call is logged for the IS-population assertions. `faults` + /// (empty for every plain `wf_*` fixture) lets the fault-attribution + /// tests below make one specific (window, params) call fail instead of + /// planting a report — the only way to reach the walk-forward stage's + /// IS-sweep / OOS-run / per-window `min_by_key` capture paths, which a + /// runner that always succeeds cannot exercise. + #[allow(clippy::type_complexity)] + struct WfFakeRunner { + log: Mutex)>>, + faults: Vec, + } + + impl WfFakeRunner { + fn new() -> Self { + WfFakeRunner { log: Mutex::new(Vec::new()), faults: Vec::new() } + } + + fn faulty(faults: Vec) -> Self { + WfFakeRunner { log: Mutex::new(Vec::new()), faults } + } + } + + impl MemberRunner for WfFakeRunner { + fn run_member( + &self, + _cell: &CellSpec, + params: &[(String, Scalar)], + window_ms: (i64, i64), + ) -> Result { + self.log.lock().unwrap().push((window_ms, params.to_vec())); + if let Some((_, _, fault)) = + self.faults.iter().find(|(w, p, _)| *w == window_ms && p == params) + { + return Err(fault.clone()); + } + let total: f64 = params + .iter() + .map(|(_, s)| match s { + Scalar::I64(v) => *v as f64, + Scalar::F64(v) => *v, + _ => 0.0, + }) + .sum(); + Ok(RunReport { + manifest: RunManifest { + commit: "wf-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: None, + topology_hash: None, + project: None, + }, + metrics: RunMetrics { + total_pips: total, + max_drawdown: 0.0, + bias_sign_flips: 0, + r: None, + }, + }) + } + } + + fn run_wf( + campaign: &CampaignDoc, + process: &ProcessDoc, + runner: &WfFakeRunner, + registry: &Registry, + ) -> Result { + let strategies = vec![("s".repeat(64), "{}".to_string())]; + let campaign_id = "e".repeat(64); + execute(&campaign_id, campaign, process, &strategies, runner, registry) + } + + #[test] + fn wf_rolls_the_declared_window_in_ms() { + let registry = wf_registry("rolls"); + let campaign = wf_campaign((0, 99)); + let process = wf_process(None, 40, 20, 20); + let runner = WfFakeRunner::new(); + let outcome = run_wf(&campaign, &process, &runner, ®istry).expect("wf executes"); + + // The oracle is the roller's own math over the same config. + let expected: Vec<_> = + WindowRoller::new((Timestamp(0), Timestamp(99)), 40, 20, 20, RollMode::Rolling) + .expect("valid roll") + .collect(); + assert_eq!(expected.len(), 3, "fixture sanity: 3 windows over 0..=99"); + + let fam = outcome.cells[0] + .families + .iter() + .find(|f| f.block == "std::walk_forward") + .expect("wf family present"); + assert_eq!(fam.reports.len(), expected.len()); + for (report, bounds) in fam.reports.iter().zip(&expected) { + assert_eq!(report.manifest.window, (bounds.oos.0, bounds.oos.1)); + } + } + + #[test] + fn wf_searches_only_the_survivor_points() { + let registry = wf_registry("survivors"); + let campaign = wf_campaign((0, 99)); + // total_pips == len; gate total_pips > 2.5 keeps len 3 and len 4 of [1,2,3,4] + let process = wf_process(Some(2.5), 40, 20, 20); + let runner = WfFakeRunner::new(); + run_wf(&campaign, &process, &runner, ®istry).expect("wf executes"); + + let is_bounds: Vec<(i64, i64)> = + WindowRoller::new((Timestamp(0), Timestamp(99)), 40, 20, 20, RollMode::Rolling) + .expect("valid roll") + .map(|w| (w.is.0 .0, w.is.1 .0)) + .collect(); + let log = runner.log.lock().unwrap(); + let mut is_total = 0; + let mut seen3 = 0; + let mut seen4 = 0; + for (window, params) in log.iter() { + if !is_bounds.contains(window) { + continue; + } + is_total += 1; + match params[0].1 { + Scalar::I64(3) => seen3 += 1, + Scalar::I64(4) => seen4 += 1, + other => panic!("gated-out point ran in an IS window: {other:?}"), + } + } + // 3 windows x exactly the 2 surviving points + assert_eq!(is_total, 6); + assert_eq!(seen3, 3); + assert_eq!(seen4, 3); + } + + #[test] + fn wf_stamps_selection_on_oos_members() { + let registry = wf_registry("selection"); + let campaign = wf_campaign((0, 99)); + let process = wf_process(None, 40, 20, 20); + let runner = WfFakeRunner::new(); + let outcome = run_wf(&campaign, &process, &runner, ®istry).expect("wf executes"); + + let fam = outcome.cells[0] + .families + .iter() + .find(|f| f.block == "std::walk_forward") + .expect("wf family present"); + assert!(!fam.reports.is_empty()); + for report in &fam.reports { + let sel = report + .manifest + .selection + .as_ref() + .expect("every OOS member carries its IS selection"); + assert_eq!(sel.selection_metric, "total_pips"); + assert_eq!(sel.seed, Some(7), "deflation is seeded from campaign.seed"); + } + } + + #[test] + fn wf_roller_refusal_maps_to_window_fault() { + let registry = wf_registry("window-fault"); + // 0..=49 cannot fit is 40 + oos 20 (window 0 needs 59) -> the roller + // refuses at runtime (zero lengths are already doc-tier faults). + let campaign = wf_campaign((0, 49)); + let process = wf_process(None, 40, 20, 20); + let runner = WfFakeRunner::new(); + let result = run_wf(&campaign, &process, &runner, ®istry); + let Err(err) = result else { panic!("span too short for one window must refuse") }; + match err { + ExecFault::Window { stage, detail } => { + assert_eq!(stage, 1, "walk_forward is pipeline stage 1 here"); + assert!( + detail.contains("SpanTooShort"), + "detail carries the roller refusal: {detail}" + ); + } + other => panic!("expected ExecFault::Window, got {other:?}"), + } + } + + /// A member fault raised INSIDE a window's in-sample sweep aborts the + /// walk_forward stage with that fault, and among several faulted windows + /// the LOWEST roll index wins (the per-window `min_by_key` attribution + /// mirrors the sweep stage's own lowest-enumeration-index convention) — + /// never the faulted stage's family, since a captured fault returns + /// before any registry write. + #[test] + fn wf_is_member_fault_lowest_window_wins() { + let registry = wf_registry("is-fault"); + let campaign = wf_campaign((0, 99)); + let process = wf_process(None, 40, 20, 20); + // IS bounds: w0 (0,39), w1 (20,59), w2 (40,79) — fault both w1's and + // w2's IS runs; the widx-1 fault must win over widx-2's. + let runner = WfFakeRunner::faulty(vec![ + ( + (20, 59), + vec![("len".to_string(), Scalar::i64(3))], + MemberFault::Run("is-w1".to_string()), + ), + ( + (40, 79), + vec![("len".to_string(), Scalar::i64(2))], + MemberFault::Run("is-w2".to_string()), + ), + ]); + let err = run_wf(&campaign, &process, &runner, ®istry) + .expect_err("a faulted IS member aborts the walk_forward stage"); + match err { + ExecFault::Member(fault) => assert_eq!( + fault, + MemberFault::Run("is-w1".to_string()), + "the LOWEST window index's fault wins, not w2's later one", + ), + other => panic!("expected ExecFault::Member, got {other:?}"), + } + let members = registry.load_family_members().expect("load members"); + assert!(members.iter().all(|m| m.kind != FamilyKind::WalkForward)); + assert!(registry.load_campaign_runs().expect("load campaign runs").is_empty()); + } + + /// A member fault raised running the IS winner over its OUT-of-sample + /// bounds (a call distinct from every IS-sweep call, since the OOS + /// window's bounds differ from every window's IS bounds) also aborts the + /// walk_forward stage with that fault, and — like the IS-sweep path — + /// never persists the faulted stage's family. + #[test] + fn wf_oos_member_fault_aborts_walk_forward() { + let registry = wf_registry("oos-fault"); + let campaign = wf_campaign((0, 99)); + let process = wf_process(None, 40, 20, 20); + // total_pips == len, so the IS winner is always len=4 (argmax over + // [1,2,3,4]); fault only window 0's OOS run of that winner (OOS + // bounds (40,59) match no window's IS bounds and no other window's + // OOS bounds). + let runner = WfFakeRunner::faulty(vec![( + (40, 59), + vec![("len".to_string(), Scalar::i64(4))], + MemberFault::Run("oos-w0".to_string()), + )]); + let err = run_wf(&campaign, &process, &runner, ®istry) + .expect_err("a faulted OOS member aborts the walk_forward stage"); + match err { + ExecFault::Member(fault) => { + assert_eq!(fault, MemberFault::Run("oos-w0".to_string())) + } + other => panic!("expected ExecFault::Member, got {other:?}"), + } + let members = registry.load_family_members().expect("load members"); + assert!(members.iter().all(|m| m.kind != FamilyKind::WalkForward)); + } + + /// select_sweep_winner's Plateau arms dispatch through the public + /// `execute()` path (only Argmax is exercised by the other wf_* tests): + /// each `SelectRule` variant lands on its matching `SelectionMode`. + #[test] + fn wf_sweep_stage_dispatches_plateau_select_modes() { + for (i, (select, expected_mode)) in [ + (SelectRule::PlateauMean, SelectionMode::PlateauMean), + (SelectRule::PlateauWorst, SelectionMode::PlateauWorst), + ] + .into_iter() + .enumerate() + { + let registry = wf_registry(&format!("plateau-{i}")); + let campaign = wf_campaign((0, 99)); + let process = ProcessDoc { + format_version: 1, + kind: DocKind::Process, + name: "wf-plateau-proc".to_string(), + description: None, + pipeline: vec![StageBlock::Sweep { + metric: "total_pips".to_string(), + select, + deflate: false, + }], + }; + let runner = WfFakeRunner::new(); + let outcome = + run_wf(&campaign, &process, &runner, ®istry).expect("plateau sweep executes"); + let sel = &outcome.cells[0].selections[0]; + assert_eq!( + sel.selection.mode, expected_mode, + "SelectRule::{select:?} must dispatch to SelectionMode::{expected_mode:?}" + ); + } + } + + /// `WfMode::Anchored` maps to the engine's `RollMode::Anchored` (only + /// Rolling is exercised by the other wf_* tests) — the oracle is the same + /// roller construction the stage itself uses, just with `RollMode::Anchored`. + #[test] + fn wf_anchored_mode_maps_to_engine_rollmode_anchored() { + let registry = wf_registry("anchored"); + let campaign = wf_campaign((0, 99)); + let process = ProcessDoc { + format_version: 1, + kind: DocKind::Process, + name: "wf-anchored-proc".to_string(), + description: None, + pipeline: vec![ + StageBlock::Sweep { + metric: "total_pips".to_string(), + select: SelectRule::Argmax, + deflate: false, + }, + StageBlock::WalkForward { + in_sample_ms: 40, + out_of_sample_ms: 20, + step_ms: 20, + mode: WfMode::Anchored, + metric: "total_pips".to_string(), + select: SelectRule::Argmax, + }, + ], + }; + let runner = WfFakeRunner::new(); + let outcome = run_wf(&campaign, &process, &runner, ®istry).expect("anchored wf executes"); + + let expected: Vec<_> = + WindowRoller::new((Timestamp(0), Timestamp(99)), 40, 20, 20, RollMode::Anchored) + .expect("valid roll") + .collect(); + assert!(!expected.is_empty(), "fixture sanity: at least one anchored window"); + + let fam = outcome.cells[0] + .families + .iter() + .find(|f| f.block == "std::walk_forward") + .expect("wf family present"); + assert_eq!(fam.reports.len(), expected.len()); + for (report, bounds) in fam.reports.iter().zip(&expected) { + assert_eq!(report.manifest.window, (bounds.oos.0, bounds.oos.1)); + } + } +} diff --git a/crates/aura-campaign/tests/execute.rs b/crates/aura-campaign/tests/execute.rs new file mode 100644 index 0000000..e795597 --- /dev/null +++ b/crates/aura-campaign/tests/execute.rs @@ -0,0 +1,360 @@ +//! 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 { + 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 { + 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) -> 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(), ®) + .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![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(), ®) + .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(), ®) + .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(), ®) + .expect("first run"); + let second = execute(CAMPAIGN_ID, &doc, &proc_doc, &strategies(), &FakeRunner::clean(), ®) + .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, ®) + .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(), ®) + .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(), ®) + .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()); +}