feat(research,campaign,cli): chunked instrument-major cell walk + duplicate-instrument refusal

Two pieces of the parallel-cell-loop iteration (spec: parallel-cell-loop),
still sequential — behaviour-identical results, suite green unchanged:

- validate_campaign refuses a campaign document listing the same instrument
  twice (DocFault::DuplicateInstrument + CLI prose). A duplicate enumerates
  two identical cells and is the one input that could make two cells share a
  registry family name; with distinct instruments the per-cell family name
  (all four cell axes + stage ordinal) is unique by construction, so the
  upcoming parallel appends can never race one name's run-index assignment.
- aura_campaign::execute flattens the 4-nested cell loop into a document-
  order-indexed cell list, groups it by instrument ordinal, and walks
  sequential chunks of K instrument groups; results land in index-addressed
  slots and the ordered outputs (outcomes, realizations, nominee groups) are
  rebuilt in document order. K arrives as a new NonZeroUsize parameter
  (default DEFAULT_PARALLEL_INSTRUMENTS = 4), exposed as
  --parallel-instruments on aura campaign run and threaded through the
  run_campaign chain; the dissolved-verb sugar paths pass the default.

The chunk walk is the structural RAM bound for the parallel flip that
follows: no cell of an instrument outside the current chunk can be in
flight at all (chosen over a semaphore gate — blocking inside pool tasks
risks worker starvation; decision log on #277). Verified: workspace suite
green unchanged, clippy -D warnings clean, E2E fixtures pin the CLI prose
and the document-order persistence across K.

refs #277

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<CampaignOutcome, ExecFault> {
     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<CellOutcome> = Vec::new();
-    let mut cells_rec: Vec<CellRealization> = Vec::new();
-    // Per-(strategy_ordinal, window_ordinal) nominees across instruments, in
-    // cell-loop (instrument) order — the campaign-scope generalize input.
-    type Nominee = Option<(Vec<(String, Scalar)>, RunReport)>;
-    let mut nominees: BTreeMap<(usize, usize, usize), Vec<(String, Nominee)>> = BTreeMap::new();
+    // 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<PlannedCell> = 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 &regimes {
-                    let cell = CellSpec {
-                        strategy_ordinal,
-                        strategy_id: strategy_id.clone(),
-                        blueprint_json: blueprint_json.clone(),
-                        axes: entry.axes.clone(),
-                        instrument: instrument.clone(),
-                        window_ms: (window.from_ms, window.to_ms),
-                        regime: *regime,
-                        regime_ordinal: *regime_ordinal,
-                    };
-                    let (outcome, realization) = run_cell(
-                        &cell,
-                        process,
-                        campaign_prefix,
+                    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<usize>> = vec![Vec::new(); campaign.data.instruments.len()];
+    for p in &planned {
+        groups[p.instrument_ordinal].push(p.doc_index);
+    }
+    let mut slots: Vec<Option<(CellOutcome, CellRealization)>> =
+        (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<CellOutcome> = Vec::with_capacity(planned.len());
+    let mut cells_rec: Vec<CellRealization> = 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.

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<CampaignOutcome, ExecFault> {
         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, &registry)
-            .expect("a member panic in a fold is contained, not a process abort");
+        let outcome = execute(
+            &"e".repeat(64),
+            &campaign,
+            &process,
+            &strategies,
+            &WfPanicRunner,
+            &registry,
+            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