diff --git a/crates/aura-campaign/src/exec.rs b/crates/aura-campaign/src/exec.rs index 0466ae9..9eaca25 100644 --- a/crates/aura-campaign/src/exec.rs +++ b/crates/aura-campaign/src/exec.rs @@ -7,6 +7,7 @@ //! in the crate's preflight, which `execute` runs before any member runs. use std::collections::BTreeMap; +use std::num::NonZeroUsize; use std::sync::Mutex; use aura_analysis::{FamilySelection, SelectionMode}; @@ -86,6 +87,7 @@ pub fn execute( strategies: &[(String, String)], runner: &dyn MemberRunner, registry: &Registry, + parallel_instruments: NonZeroUsize, ) -> Result { preflight(process, campaign)?; if strategies.len() != campaign.strategies.len() { @@ -115,12 +117,15 @@ pub fn execute( } let campaign_prefix = &campaign_id[..8]; - let mut cells_out: Vec = Vec::new(); - let mut cells_rec: Vec = Vec::new(); - // Per-(strategy_ordinal, window_ordinal) nominees across instruments, in - // cell-loop (instrument) order — the campaign-scope generalize input. - type Nominee = Option<(Vec<(String, Scalar)>, RunReport)>; - let mut nominees: BTreeMap<(usize, usize, usize), Vec<(String, Nominee)>> = BTreeMap::new(); + // Flatten: the same 4-level nesting enumerates every cell with its + // document-order index. Execution order below is instrument-major and + // chunked; RESULT order stays this document order (C1). + struct PlannedCell { + doc_index: usize, + instrument_ordinal: usize, + window_ordinal: usize, + cell: CellSpec, + } // Absent/empty risk = a single default cell (regime None, ordinal 0); a // non-empty list maps each regime to Some. The default's value is the member // runner's — the doc is never mutated (its content id hashes with risk absent). @@ -129,42 +134,77 @@ pub fn execute( } else { campaign.risk.iter().copied().enumerate().map(|(i, r)| (i, Some(r))).collect() }; + let mut planned: 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 (instrument_ordinal, instrument) in campaign.data.instruments.iter().enumerate() { for (window_ordinal, window) in campaign.data.windows.iter().enumerate() { for (regime_ordinal, regime) in ®imes { - let cell = CellSpec { - strategy_ordinal, - strategy_id: strategy_id.clone(), - blueprint_json: blueprint_json.clone(), - axes: entry.axes.clone(), - instrument: instrument.clone(), - window_ms: (window.from_ms, window.to_ms), - regime: *regime, - regime_ordinal: *regime_ordinal, - }; - let (outcome, realization) = run_cell( - &cell, - process, - campaign_prefix, + planned.push(PlannedCell { + doc_index: planned.len(), + instrument_ordinal, window_ordinal, - campaign.seed, - runner, - registry, - )?; - nominees - .entry((strategy_ordinal, window_ordinal, *regime_ordinal)) - .or_default() - .push((instrument.clone(), outcome.nominee.clone())); - cells_out.push(outcome); - cells_rec.push(realization); + cell: CellSpec { + strategy_ordinal, + strategy_id: strategy_id.clone(), + blueprint_json: blueprint_json.clone(), + axes: entry.axes.clone(), + instrument: instrument.clone(), + window_ms: (window.from_ms, window.to_ms), + regime: *regime, + regime_ordinal: *regime_ordinal, + }, + }); } } } } + // Group by instrument ordinal (not name: correct even on unvalidated + // duplicate-name input), then walk sequential chunks of K groups — the + // structural residency bound: no cell of an instrument outside the + // current chunk can be in flight at all. + let mut groups: Vec> = vec![Vec::new(); campaign.data.instruments.len()]; + for p in &planned { + groups[p.instrument_ordinal].push(p.doc_index); + } + let mut slots: Vec> = + (0..planned.len()).map(|_| None).collect(); + for chunk in groups.chunks(parallel_instruments.get()) { + for &doc_index in chunk.iter().flatten() { + let p = &planned[doc_index]; + let pair = run_cell( + &p.cell, + process, + campaign_prefix, + p.window_ordinal, + campaign.seed, + runner, + registry, + )?; + slots[doc_index] = Some(pair); + } + } + + // Rebuild the ordered outputs in document order — identical pushes to the + // former nested loop, so nominee grouping and record layout are unchanged. + let mut cells_out: Vec = Vec::with_capacity(planned.len()); + let mut cells_rec: Vec = Vec::with_capacity(planned.len()); + // Per-(strategy_ordinal, window_ordinal) nominees across instruments, in + // document (instrument) order — the campaign-scope generalize input. + type Nominee = Option<(Vec<(String, Scalar)>, RunReport)>; + let mut nominees: BTreeMap<(usize, usize, usize), Vec<(String, Nominee)>> = BTreeMap::new(); + for (p, slot) in planned.iter().zip(slots) { + let (outcome, realization) = slot.expect("the walk returned early on any fault"); + nominees + .entry((p.cell.strategy_ordinal, p.window_ordinal, p.cell.regime_ordinal)) + .or_default() + .push((p.cell.instrument.clone(), outcome.nominee.clone())); + cells_out.push(outcome); + cells_rec.push(realization); + } + // Campaign-scope generalize (#200 d2): for each (strategy, window), grade // the per-instrument nominees via the shipped `generalization`; fewer than // 2 winners is a recorded shortfall, never computed around. diff --git a/crates/aura-campaign/src/lib.rs b/crates/aura-campaign/src/lib.rs index b430738..4d72291 100644 --- a/crates/aura-campaign/src/lib.rs +++ b/crates/aura-campaign/src/lib.rs @@ -17,12 +17,24 @@ pub use exec::{ }; use std::collections::BTreeMap; +use std::num::NonZeroUsize; use aura_core::Scalar; use aura_engine::RunReport; use aura_registry::{check_r_metric, RegistryError}; use aura_research::{Axis, CampaignDoc, Cmp, ProcessDoc, RiskRegime, SelectRule, StageBlock}; +/// Default bound on distinct concurrently-active instruments in the campaign +/// cell loop (the RAM lever): the external data-server retains an +/// instrument's parsed files while any of its cells is live (~1.4 GB per +/// instrument on the reference dataset), so the resident set scales with this +/// bound, not with pool workers. 1 reproduces the sequential loop's +/// footprint. +pub const DEFAULT_PARALLEL_INSTRUMENTS: NonZeroUsize = match NonZeroUsize::new(4) { + Some(n) => n, + None => panic!("4 is nonzero"), +}; + /// One structural cell of the campaign matrix: (strategy, instrument, /// window) — #198 decision 7. pub struct CellSpec { @@ -1142,7 +1154,15 @@ mod wf_tests { ) -> Result { let strategies = vec![("s".repeat(64), "{}".to_string())]; let campaign_id = "e".repeat(64); - execute(&campaign_id, campaign, process, &strategies, runner, registry) + execute( + &campaign_id, + campaign, + process, + &strategies, + runner, + registry, + crate::DEFAULT_PARALLEL_INSTRUMENTS, + ) } #[test] @@ -1399,8 +1419,16 @@ mod wf_tests { let campaign = wf_campaign((0, 99)); let process = wf_process(None, 40, 20, 20); let strategies = vec![("s".repeat(64), "{}".to_string())]; - let outcome = execute(&"e".repeat(64), &campaign, &process, &strategies, &WfPanicRunner, ®istry) - .expect("a member panic in a fold is contained, not a process abort"); + let outcome = execute( + &"e".repeat(64), + &campaign, + &process, + &strategies, + &WfPanicRunner, + ®istry, + crate::DEFAULT_PARALLEL_INSTRUMENTS, + ) + .expect("a member panic in a fold is contained, not a process abort"); let cell = &outcome.record.cells[0]; let wf_stage = cell diff --git a/crates/aura-campaign/tests/execute.rs b/crates/aura-campaign/tests/execute.rs index 68d6b40..514c103 100644 --- a/crates/aura-campaign/tests/execute.rs +++ b/crates/aura-campaign/tests/execute.rs @@ -4,7 +4,9 @@ use std::collections::BTreeMap; -use aura_campaign::{execute, CellSpec, ExecFault, MemberFault, MemberRunner}; +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, StageBootstrap}; @@ -247,7 +249,7 @@ 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(), ®) + let out = execute(CAMPAIGN_ID, &doc, &proc_doc, &strategies(), &FakeRunner::clean(), ®, DEFAULT_PARALLEL_INSTRUMENTS) .expect("sweep-only campaign executes"); // outcome payloads: one cell, one family of 4 members, one selection @@ -310,7 +312,7 @@ 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(), ®) + let out = execute(CAMPAIGN_ID, &doc, &proc_doc, &strategies(), &FakeRunner::clean(), ®, DEFAULT_PARALLEL_INSTRUMENTS) .expect("selection-free sweep campaign executes"); // outcome payloads: one cell, one family of 4 members, NO selection @@ -347,7 +349,7 @@ 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(), ®) + let out = execute(CAMPAIGN_ID, &doc, &proc_doc, &strategies(), &FakeRunner::clean(), ®, 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 @@ -366,7 +368,7 @@ fn execute_zero_survivors_truncates_cell_and_continues() { 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(), ®) + let out = execute(CAMPAIGN_ID, &doc, &proc_doc, &strategies(), &FakeRunner::clean(), ®, 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"); @@ -383,14 +385,46 @@ fn execute_zero_survivors_truncates_cell_and_continues() { 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(), ®, 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!["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(), ®) + let first = execute(CAMPAIGN_ID, &doc, &proc_doc, &strategies(), &FakeRunner::clean(), ®, DEFAULT_PARALLEL_INSTRUMENTS) .expect("first run"); - let second = execute(CAMPAIGN_ID, &doc, &proc_doc, &strategies(), &FakeRunner::clean(), ®) + let second = execute(CAMPAIGN_ID, &doc, &proc_doc, &strategies(), &FakeRunner::clean(), ®, DEFAULT_PARALLEL_INSTRUMENTS) .expect("second run"); // run counter advances per campaign id @@ -437,7 +471,7 @@ fn execute_member_fault_records_a_failed_cell_and_continues() { let runner = FakeRunner { faults: vec![(point(2, 9), MemberFault::Run("boom".to_string()))], }; - let out = execute(CAMPAIGN_ID, &doc, &proc_doc, &strategies(), &runner, ®) + let out = execute(CAMPAIGN_ID, &doc, &proc_doc, &strategies(), &runner, ®, 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 { @@ -460,7 +494,7 @@ fn execute_member_panic_is_contained_as_a_failed_cell() { 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, ®) + let out = execute(CAMPAIGN_ID, &doc, &proc_doc, &strategies(), &runner, ®, DEFAULT_PARALLEL_INSTRUMENTS) .expect("a member panic is contained, not a process abort"); assert!(out.record.cells.iter().all(|c| c .fault @@ -474,7 +508,7 @@ 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(), ®) + let out = execute(CAMPAIGN_ID, &doc, &proc_doc, &strategies(), &FakeRunner::clean(), ®, 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); @@ -488,7 +522,7 @@ 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(), ®) + let err = execute("short", &doc, &proc_doc, &strategies(), &FakeRunner::clean(), ®, DEFAULT_PARALLEL_INSTRUMENTS) .expect_err("a non-64-hex campaign id is refused, not sliced"); match err { ExecFault::PipelineShape { detail } => assert!( @@ -509,7 +543,7 @@ fn execute_mc_after_gate_bootstraps_each_survivor() { gate_stage(0.3), StageBlock::MonteCarlo { resamples: 200, block_len: 2 }, ]); - let out = execute(CAMPAIGN_ID, &doc, &proc_doc, &strategies(), &FakeRunner::clean(), ®) + let out = execute(CAMPAIGN_ID, &doc, &proc_doc, &strategies(), &FakeRunner::clean(), ®, DEFAULT_PARALLEL_INSTRUMENTS) .expect("mc-after-gate campaign executes"); // planted nets .26/.29/.36/.39 -> gate gt 0.3 keeps ordinals 2 and 3 @@ -550,7 +584,7 @@ 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(), ®) + let out = execute(CAMPAIGN_ID, &doc, &proc_doc, &strategies(), &FakeRunner::clean(), ®, DEFAULT_PARALLEL_INSTRUMENTS) .expect("grid-then-wf campaign executes"); // std::grid persists no family of its own — only std::walk_forward's does @@ -590,7 +624,7 @@ fn execute_mc_after_wf_pools_the_oos_series() { wf_stage(), StageBlock::MonteCarlo { resamples: 200, block_len: 3 }, ]); - let out = execute(CAMPAIGN_ID, &doc, &proc_doc, &strategies(), &FakeRunner::clean(), ®) + let out = execute(CAMPAIGN_ID, &doc, &proc_doc, &strategies(), &FakeRunner::clean(), ®, DEFAULT_PARALLEL_INSTRUMENTS) .expect("mc-after-wf campaign executes"); // the roller yields 2 windows; both pick the planted argmax (3,9) @@ -626,7 +660,7 @@ fn execute_mc_zero_trade_member_records_degenerate() { 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, ®) + let out = execute(CAMPAIGN_ID, &doc, &proc_doc, &strategies(), &runner, ®, DEFAULT_PARALLEL_INSTRUMENTS) .expect("a zero-trade member is a valid result, not a fault"); let mc = &out.record.cells[0].stages[1]; @@ -649,7 +683,7 @@ fn execute_generalize_across_two_instruments() { // 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(), ®) + let out = execute(CAMPAIGN_ID, &doc, &proc_doc, &strategies(), &FakeRunner::clean(), ®, DEFAULT_PARALLEL_INSTRUMENTS) .expect("generalize campaign executes"); assert_eq!(out.record.generalizations.len(), 1); @@ -708,7 +742,7 @@ fn execute_generalize_shortfall_records_missing() { // 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(), ®) + let out = execute(CAMPAIGN_ID, &doc, &proc_doc, &strategies(), &FakeRunner::clean(), ®, 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"); @@ -729,7 +763,7 @@ 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(), ®) + let out = execute(CAMPAIGN_ID, &doc, &proc_doc, &strategies(), &FakeRunner::clean(), ®, DEFAULT_PARALLEL_INSTRUMENTS) .expect("sweep-only generalize campaign executes"); // no wf ran: each cell nominates its sweep winner (planted argmax (3,9)) @@ -781,7 +815,7 @@ fn execute_mc_and_generalize_compose_in_one_pipeline() { StageBlock::MonteCarlo { resamples: 200, block_len: 2 }, generalize_stage(), ]); - let out = execute(CAMPAIGN_ID, &doc, &proc_doc, &strategies(), &FakeRunner::clean(), ®) + let out = execute(CAMPAIGN_ID, &doc, &proc_doc, &strategies(), &FakeRunner::clean(), ®, DEFAULT_PARALLEL_INSTRUMENTS) .expect("full v2 pipeline executes"); // both cells realize all four per-cell stages; generalize contributes no @@ -839,7 +873,7 @@ fn execute_persist_taps_stamps_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(), ®) + let out = execute(CAMPAIGN_ID, &doc, &proc_doc, &strategies(), &FakeRunner::clean(), ®, DEFAULT_PARALLEL_INSTRUMENTS) .expect("persist_taps campaign executes"); // returned record: the derived name over the first run counter @@ -857,7 +891,7 @@ fn execute_persist_taps_stamps_trace_name() { 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(), ®) + let second = execute(CAMPAIGN_ID, &doc, &proc_doc, &strategies(), &FakeRunner::clean(), ®, 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())); @@ -876,7 +910,7 @@ fn execute_runs_one_family_per_regime_with_distinct_ids() { 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(), ®) + let out = execute(CAMPAIGN_ID, &doc, &proc_doc, &strategies(), &FakeRunner::clean(), ®, DEFAULT_PARALLEL_INSTRUMENTS) .expect("two-regime campaign executes"); // one cell per regime, each with its own sweep family. @@ -903,7 +937,7 @@ fn execute_generalize_keeps_regimes_separate() { 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(), ®) + let out = execute(CAMPAIGN_ID, &doc, &proc_doc, &strategies(), &FakeRunner::clean(), ®, DEFAULT_PARALLEL_INSTRUMENTS) .expect("two-regime generalize campaign executes"); // 2 instruments x 2 regimes -> 4 cells, but exactly 2 generalize groups diff --git a/crates/aura-cli/src/campaign_run.rs b/crates/aura-cli/src/campaign_run.rs index a743c76..8e09880 100644 --- a/crates/aura-cli/src/campaign_run.rs +++ b/crates/aura-cli/src/campaign_run.rs @@ -15,6 +15,7 @@ //! visible to child modules without promotion. use std::collections::{BTreeMap, HashSet}; +use std::num::NonZeroUsize; use std::path::{Path, PathBuf}; use std::sync::{mpsc, Arc}; @@ -478,7 +479,11 @@ pub(crate) enum RunPresentation { /// is a refusal `campaign_cmd` renders as `aura: {msg}` + exit 1; an `Ok` /// carries the failed-cell count (#272) so the caller can exit 3 on a /// completed-with-failures run via `exit_on_campaign_result`. -pub(crate) fn run_campaign(target: &str, env: &Env) -> Result { +pub(crate) fn run_campaign( + target: &str, + env: &Env, + parallel_instruments: NonZeroUsize, +) -> Result { // Project gate FIRST: nothing (not even the file-sugar registration) // touches a store outside a project. if env.provenance().is_none() { @@ -511,7 +516,7 @@ pub(crate) fn run_campaign(target: &str, env: &Env) -> Result { )); }; - run_campaign_by_id(&campaign_id, env, RunPresentation::Full) + run_campaign_by_id(&campaign_id, env, RunPresentation::Full, parallel_instruments) } /// An executed campaign plus the context its presenter and the dissolved-verb @@ -540,8 +545,9 @@ pub(crate) fn run_campaign_by_id( campaign_id: &str, env: &Env, presentation: RunPresentation, + parallel_instruments: NonZeroUsize, ) -> Result { - let run = run_campaign_returning(campaign_id, env)?; + let run = run_campaign_returning(campaign_id, env, parallel_instruments)?; present_campaign(run, presentation, env) } @@ -556,6 +562,7 @@ pub(crate) fn run_campaign_by_id( pub(crate) fn run_campaign_returning( campaign_id: &str, env: &Env, + parallel_instruments: NonZeroUsize, ) -> Result { let registry = env.registry(); let campaign_text = registry @@ -635,6 +642,7 @@ pub(crate) fn run_campaign_returning( &strategies, &runner, ®istry, + parallel_instruments, ) .map_err(|f| exec_fault_prose(&f))?; diff --git a/crates/aura-cli/src/research_docs.rs b/crates/aura-cli/src/research_docs.rs index 520f4ab..7c68e64 100644 --- a/crates/aura-cli/src/research_docs.rs +++ b/crates/aura-cli/src/research_docs.rs @@ -137,6 +137,9 @@ pub(crate) fn doc_fault_prose(f: &DocFault) -> String { format!("pipeline[{stage}]: walk_forward {field} must be > 0") } DocFault::EmptyInstruments => "data.instruments is empty".into(), + DocFault::DuplicateInstrument { index, instrument } => { + format!("data.instruments[{index}]: \"{instrument}\" is listed more than once") + } DocFault::NoWindow => "data.windows is empty".into(), DocFault::BadWindow { index } => { format!("data.windows[{index}]: from_ms must be earlier than to_ms") @@ -313,7 +316,13 @@ pub enum CampaignSub { Register { file: PathBuf }, /// Execute a stored campaign into a realized run-set (a .json file is /// register-then-run sugar; the canonical address is the content id). - Run { target: String }, + Run { + target: String, + /// Bound on distinct instruments resident in parallel (the RAM + /// lever; 1 reproduces the sequential loop's footprint). + #[arg(long, default_value_t = aura_campaign::DEFAULT_PARALLEL_INSTRUMENTS)] + parallel_instruments: std::num::NonZeroUsize, + }, /// List stored campaign realizations, or dump one campaign's records /// (the bare store lines, not the run-emit wrapper). Runs { campaign: Option, run: Option }, @@ -420,8 +429,12 @@ fn campaign_runs(campaign: Option<&str>, run: Option, env: &Env) -> Resul /// via `exit_on_campaign_result`), so it is pulled out of the unified /// `Result<(), String>` match the other four subcommands still share. pub fn campaign_cmd(cmd: CampaignCmd, env: &Env) { - if let CampaignSub::Run { target } = &cmd.sub { - crate::exit_on_campaign_result(crate::campaign_run::run_campaign(target, env)); + if let CampaignSub::Run { target, parallel_instruments } = &cmd.sub { + crate::exit_on_campaign_result(crate::campaign_run::run_campaign( + target, + env, + *parallel_instruments, + )); return; } let result = match &cmd.sub { diff --git a/crates/aura-cli/src/verb_sugar.rs b/crates/aura-cli/src/verb_sugar.rs index b57e450..ed58b05 100644 --- a/crates/aura-cli/src/verb_sugar.rs +++ b/crates/aura-cli/src/verb_sugar.rs @@ -345,7 +345,12 @@ pub(crate) fn run_sweep_sugar( validate_before_register(&generated.process, &generated.campaign, inv.blueprint_canonical, &probe_params, env)?; let reg = env.registry(); let (_process_id, campaign_id) = register_generated(®, &generated)?; - crate::campaign_run::run_campaign_by_id(&campaign_id, env, crate::campaign_run::RunPresentation::MemberLinesOnly) + crate::campaign_run::run_campaign_by_id( + &campaign_id, + env, + crate::campaign_run::RunPresentation::MemberLinesOnly, + aura_campaign::DEFAULT_PARALLEL_INSTRUMENTS, + ) } /// Build the `CrossInstrument` family members from an executed generalize @@ -386,7 +391,11 @@ pub(crate) fn run_generalize_sugar( validate_before_register(&generated.process, &generated.campaign, inv.blueprint_canonical, &probe_params, env)?; let reg = env.registry(); let (_process_id, campaign_id) = register_generated_g(®, &generated)?; - let run = crate::campaign_run::run_campaign_returning(&campaign_id, env)?; + let run = crate::campaign_run::run_campaign_returning( + &campaign_id, + env, + aura_campaign::DEFAULT_PARALLEL_INSTRUMENTS, + )?; // Reprint the verb's aggregate line from the recorded cross-instrument // grade — byte-identical to the retired welded path's `generalize_json(&agg)`. @@ -508,7 +517,11 @@ pub(crate) fn run_walkforward_sugar( validate_before_register(&generated.process, &generated.campaign, inv.blueprint_canonical, &probe_params, env)?; let reg = env.registry(); let (_process_id, campaign_id) = register_generated_wf(®, &generated)?; - let run = crate::campaign_run::run_campaign_returning(&campaign_id, env)?; + let run = crate::campaign_run::run_campaign_returning( + &campaign_id, + env, + aura_campaign::DEFAULT_PARALLEL_INSTRUMENTS, + )?; if let Some(f) = run.outcome.record.cells.iter().find_map(|c| c.fault.as_ref()) { eprintln!("aura: walkforward cell failed at stage {}: {}", f.stage, f.detail); @@ -668,7 +681,11 @@ pub(crate) fn run_mc_sugar( validate_before_register(&generated.process, &generated.campaign, inv.blueprint_canonical, &probe_params, env)?; let reg = env.registry(); let (_process_id, campaign_id) = register_generated_mc(®, &generated)?; - let run = crate::campaign_run::run_campaign_returning(&campaign_id, env)?; + let run = crate::campaign_run::run_campaign_returning( + &campaign_id, + env, + aura_campaign::DEFAULT_PARALLEL_INSTRUMENTS, + )?; if let Some(f) = run.outcome.record.cells.iter().find_map(|c| c.fault.as_ref()) { eprintln!("aura: mc cell failed at stage {}: {}", f.stage, f.detail); diff --git a/crates/aura-cli/tests/research_docs.rs b/crates/aura-cli/tests/research_docs.rs index 49f5ed1..48fde37 100644 --- a/crates/aura-cli/tests/research_docs.rs +++ b/crates/aura-cli/tests/research_docs.rs @@ -449,6 +449,31 @@ fn campaign_validate_refuses_empty_axis_prose_exit_1() { assert!(!out.contains("EmptyAxis"), "Debug leak: {out}"); } +/// Property (#277 iteration, aura-research's `validate_campaign`): a campaign +/// whose `data.instruments` names the same instrument twice is refused at the +/// intrinsic validate tier, naming the repeated entry's index and value — not +/// silently accepted into a cell loop that would then double-run and +/// double-persist that instrument's cells under two "distinct" cell indices. +/// Data-free (no project needed): the intrinsic tier runs before any +/// referential or executor check. Exercised end to end through the built +/// binary (`aura campaign validate`), not just `validate_campaign` in +/// isolation, so a regression that dropped the CLI-side wiring of the new +/// `DocFault` variant into `doc_fault_prose` would also be caught here. +#[test] +fn campaign_validate_refuses_duplicate_instrument_prose_exit_1() { + let dir = temp_cwd("campaign-validate-duplicate-instrument"); + let bad = + CAMPAIGN_DOC.replacen(r#""instruments": ["GER40"]"#, r#""instruments": ["GER40", "GER40"]"#, 1); + write_doc(&dir, "bad.campaign.json", &bad); + let (out, code) = run_code_in(&dir, &["campaign", "validate", "bad.campaign.json"]); + assert_eq!(code, Some(1), "stdout/stderr: {out}"); + assert!( + out.contains(r#"data.instruments[1]: "GER40" is listed more than once"#), + "stdout/stderr: {out}" + ); + assert!(!out.contains("DuplicateInstrument"), "Debug leak: {out}"); +} + /// #210 risk-regime axis: a campaign document carrying a `risk` section /// round-trips through the CLI's actual `campaign validate` binary path /// (parse -> `validate_campaign`) without a fault. Before this iteration the @@ -1632,6 +1657,76 @@ fn campaign_run_synthetic_e2e_grid_then_wf_persists_no_sweep_family() { assert!(!stages[1]["family_id"].is_null(), "std::walk_forward persists its family: {stages}"); } +/// Property (#277: the flatten/group/chunk cell-loop rework): the persisted +/// `campaign_run.cells` order is DOCUMENT order (the `data.instruments` +/// declaration order) regardless of `--parallel-instruments`. The executor +/// now groups planned cells by instrument ordinal and walks sequential +/// chunks of K instrument-groups (K = `--parallel-instruments`) so at most K +/// instruments are ever resident at once, but that is purely an EXECUTION +/// schedule — the final `cells` vector is rebuilt by re-zipping each +/// execution slot back onto its original document-order index. A regression +/// that let the chunked/grouped execution order leak into the persisted +/// order (or that made the answer depend on K) would silently break every +/// downstream consumer keyed by cell index: nominee grouping +/// (`(strategy_ordinal, window_ordinal, regime_ordinal)` -> per-instrument +/// nominees), `reproduce`, and any UI rendering cells positionally. The +/// fixture deliberately lists instruments in REVERSE alphabetical order +/// (`SYMB` before `SYMA`) so document order and instrument-ordinal-ascending +/// order coincide (they always do, by construction) while a naive +/// "output = the order cells finished running" bug would still be caught by +/// asserting the concrete SYMB-before-SYMA position. K=1 (each instrument its +/// own chunk, fully sequential residency) and K=2 (both instruments resident +/// together) both must produce the identical cells[0]=SYMB, cells[1]=SYMA +/// order. Hostless: runs over the synthetic SYMA+SYMB archive. +#[test] +fn campaign_run_synthetic_e2e_cell_order_is_document_order_under_parallel_instruments() { + let (dir, _fixture) = fresh_project_with_data(); + let runs_dir = dir.join("runs"); + std::fs::remove_dir_all(&runs_dir).ok(); + let _cleanup = ScratchGuard(vec![ + ScratchPath::Dir(runs_dir.clone()), + ScratchPath::File(dir.join("cellorder.process.json")), + ScratchPath::File(dir.join("cellorder.campaign.json")), + ]); + let bp_id = seed_blueprint(&dir, "campaign-run-cellorder-seed"); + let proc_id = register_process_doc(&dir, "cellorder.process.json", SWEEP_ONLY_PROCESS_DOC); + // The window lies inside both SYMA's (2024-01..08) and SYMB's (2024-03..06) + // synthetic span — the grid-then-wf sibling test's window. + let doc = campaign_doc_json_two( + ("SYMB", "SYMA"), + "", + &bp_id, + &proc_id, + (1709251200000, 1719791999999), + ); + write_doc(&dir, "cellorder.campaign.json", &doc); + + for k in ["1", "2"] { + let (out, code) = run_code_in( + &dir, + &["campaign", "run", "cellorder.campaign.json", "--parallel-instruments", k], + ); + assert_eq!(code, Some(0), "K={k}: campaign run failed: {out}"); + let line = out + .lines() + .find(|l| l.starts_with("{\"campaign_run\":")) + .expect("the always-on final campaign_run line"); + let v: serde_json::Value = serde_json::from_str(line).expect("record parses"); + let cells = v["campaign_run"]["cells"].as_array().expect("cells array"); + assert_eq!(cells.len(), 2, "K={k}: one cell per instrument: {line}"); + assert_eq!( + cells[0]["instrument"].as_str(), + Some("SYMB"), + "K={k}: cell 0 is the FIRST document-order instrument: {line}" + ); + assert_eq!( + cells[1]["instrument"].as_str(), + Some("SYMA"), + "K={k}: cell 1 is the SECOND document-order instrument: {line}" + ); + } +} + /// The v2 boundary, campaign side: a generalize-bearing process is an /// executable shape, but `std::generalize` needs >= 2 instruments in the /// campaign — a STATIC preflight fact (the referential gate has run, no data diff --git a/crates/aura-research/src/lib.rs b/crates/aura-research/src/lib.rs index 42ad605..6b47767 100644 --- a/crates/aura-research/src/lib.rs +++ b/crates/aura-research/src/lib.rs @@ -800,6 +800,7 @@ pub enum DocFault { ZeroWalkForwardLength { stage: usize, field: &'static str }, // campaign side EmptyInstruments, + DuplicateInstrument { index: usize, instrument: String }, NoWindow, BadWindow { index: usize }, BadRegime { index: usize }, @@ -888,6 +889,15 @@ pub fn validate_campaign(doc: &CampaignDoc) -> Vec { if doc.data.instruments.is_empty() { faults.push(DocFault::EmptyInstruments); } + let mut seen_instruments = std::collections::BTreeSet::new(); + for (index, instrument) in doc.data.instruments.iter().enumerate() { + if !seen_instruments.insert(instrument.as_str()) { + faults.push(DocFault::DuplicateInstrument { + index, + instrument: instrument.clone(), + }); + } + } if doc.data.windows.is_empty() { faults.push(DocFault::NoWindow); } @@ -1989,6 +1999,19 @@ mod tests { assert!(validate_campaign(&emit).contains(&DocFault::UnknownEmitKind("pie_chart".into()))); } + #[test] + fn a_campaign_with_duplicate_instruments_is_refused() { + let ok = parse_campaign(CAMPAIGN_FIXTURE).unwrap(); + assert_eq!(validate_campaign(&ok), Vec::new()); + + let mut dup = ok.clone(); + let first = dup.data.instruments[0].clone(); + dup.data.instruments.push(first.clone()); + let index = dup.data.instruments.len() - 1; + assert!(validate_campaign(&dup) + .contains(&DocFault::DuplicateInstrument { index, instrument: first })); + } + /// #201 decision 1 (cycle 0109): the tap namespace is a CLOSED vocabulary — /// the wrap convention's four persisted sink names. A genuinely new /// observable is a new vocabulary entry (or an authored blueprint sink),