feat(campaign): parallel cell loop on the shared rayon pool

The flip: within each K-instrument chunk (the structural residency bound
from the previous commit), all cells run concurrently on rayon's process-
global pool — the same pool the engine's member/window fan-out already
shares, so cells x members x windows feed one work-stealing scheduler and
the per-cell serial seams (ingest transpose, single-threaded bootstrap)
overlap with other cells' work. Results land in index-addressed slots and
the ordered outputs are rebuilt in document order: byte-identical outcomes
across worker counts and bounds (C1).

Fault routing, restructured for a loop that cannot mid-iterate abort:
member/window faults stay #272-contained inside run_cell; non-containable
faults (incl. registry writes — infrastructure, not cell pathology) latch
an atomic abort flag so unstarted cells never run, and after the join the
lowest document-order fault among completed cells propagates as the run's
error. Nothing partial is persisted at run level on the fatal path.

Property pins: determinism across thread counts and bounds (1-thread/K=1
== 8-thread/K=2), distinct live instruments never exceed K (refcounting
stub runner, 8-thread pool), a dead family store is run-fatal with no
partial campaign-run record, member faults stay contained under the
parallel loop. E2E fixtures (previous commit) pin the CLI prose and
document-order persistence across K.

Verified: workspace suite green, clippy -D warnings clean. rayon enters
aura-campaign under the amended C16 per-case policy (comment in
Cargo.toml, mirroring aura-engine).

closes #277

crates/aura-campaign/Cargo.toml

@@ -23,3 +23,8 @@ aura-research = { path = "../aura-research" }
 # realization payloads derive serde (the registry per-case policy, INDEX.md).
 serde = { workspace = true, features = ["derive"] }
 serde_json = { workspace = true }
+# rayon is admitted under the amended C16 per-case dependency policy (INDEX.md):
+# the campaign cell loop fans C1-disjoint cells into the same process-global
+# work-stealing pool the engine's run_indexed already uses — parallelism across
+# sims, never within one. Direct versioned dep, mirroring aura-engine.
+rayon = "1"

crates/aura-campaign/src/exec.rs

@@ -8,6 +8,7 @@
 
 use std::collections::BTreeMap;
 use std::num::NonZeroUsize;
+use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::Mutex;
 
 use aura_analysis::{FamilySelection, SelectionMode};
@@ -23,6 +24,7 @@ use aura_registry::{
     PlateauMode, Registry, StageBootstrap, StageRealization, StageSelection,
 };
 use aura_research::{Axis, CampaignDoc, DocRef, ProcessDoc, SelectRule, StageBlock, WfMode};
+use rayon::prelude::*;
 
 use crate::{
     member_metric, predicate_holds, preflight, CellSpec, ExecFault, MemberFault, MemberRunner,
@@ -73,10 +75,18 @@ pub struct CampaignOutcome {
 }
 
 /// 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
+/// (strategy, instrument, window) cell. Cells are grouped by instrument
+/// ordinal and walked in sequential chunks of `parallel_instruments` groups;
+/// within a chunk, every cell runs concurrently via `rayon::par_iter` (#277:
+/// parallelism now also spans cells, bounded by `parallel_instruments` — C1
+/// still holds because cells are disjoint work, and RESULT order is rebuilt
+/// in document order regardless of completion order). A non-containable
+/// fault (see `contain`) latches a shared abort flag and short-circuits the
+/// run deterministically: propagation picks the lowest document-order fault
+/// among the cells that completed before the latch, which cell that is being
+/// scheduling-dependent but acceptable for these external, non-input faults.
+/// `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.
@@ -117,6 +127,10 @@ pub fn execute(
     }
     let campaign_prefix = &campaign_id[..8];
 
+    // A slot's per-cell outcome — the same type `run_cell` returns, named here
+    // so the chunk-walk bindings below don't nest it raw (clippy::type_complexity).
+    type CellResult = Result<(CellOutcome, CellRealization), ExecFault>;
+
     // 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).
@@ -169,21 +183,47 @@ pub fn execute(
     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();
+    let abort = AtomicBool::new(false);
+    let mut slots: Vec<Option<CellResult>> = (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);
+        let chunk_cells: Vec<usize> = chunk.iter().flatten().copied().collect();
+        let results: Vec<(usize, Option<CellResult>)> = chunk_cells
+            .par_iter()
+            .map(|&doc_index| {
+                if abort.load(Ordering::Relaxed) {
+                    // Never started: a fatal fault elsewhere already ends
+                    // the run; this slot stays empty.
+                    return (doc_index, None);
+                }
+                let p = &planned[doc_index];
+                let result = run_cell(
+                    &p.cell,
+                    process,
+                    campaign_prefix,
+                    p.window_ordinal,
+                    campaign.seed,
+                    runner,
+                    registry,
+                );
+                if result.is_err() {
+                    abort.store(true, Ordering::Relaxed);
+                }
+                (doc_index, Some(result))
+            })
+            .collect();
+        for (doc_index, result) in results {
+            slots[doc_index] = result;
+        }
+    }
+
+    // Non-containable faults stay run-fatal: propagate the lowest
+    // document-order fault among completed cells (deterministic attribution;
+    // which cells completed before the abort latched is scheduling-dependent,
+    // acceptable for external, non-input faults — see the spec).
+    for slot in slots.iter_mut() {
+        if matches!(slot, Some(Err(_))) {
+            let Some(Err(fault)) = slot.take() else { unreachable!() };
+            return Err(fault);
         }
     }
 
@@ -196,7 +236,12 @@ pub fn execute(
     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");
+        let (outcome, realization) = match slot {
+            Some(Ok(pair)) => pair,
+            // No fatal fault (checked above) means abort never latched, so
+            // every cell ran to Ok.
+            _ => unreachable!("no fatal fault, so every cell completed"),
+        };
         nominees
             .entry((p.cell.strategy_ordinal, p.window_ordinal, p.cell.regime_ordinal))
             .or_default()

crates/aura-campaign/tests/execute.rs

@@ -9,7 +9,7 @@ use aura_campaign::{
 };
 use aura_core::{Scalar, ScalarKind, Timestamp};
 use aura_engine::{r_bootstrap, RMetrics, RunManifest, RunMetrics, RunReport, SelectionMode};
-use aura_registry::{generalization, FamilyKind, Registry, StageBootstrap};
+use aura_registry::{generalization, FamilyKind, Registry, RegistryError, StageBootstrap};
 use aura_research::{
     Axis, CampaignDoc, Cmp, DataSection, DocKind, DocRef, Predicate, Presentation, ProcessDoc,
     ProcessRef, SelectRule, StageBlock, StrategyEntry, SweepSelection, WfMode, Window,
@@ -220,6 +220,22 @@ fn generalize_stage() -> StageBlock {
     StageBlock::Generalize { metric: "net_expectancy_r".to_string() }
 }
 
+/// A registry whose family store path is pre-occupied by a directory:
+/// `append_family`'s `OpenOptions::open` then fails with a real IO error
+/// (`ErrorKind::IsADirectory`) — a `RegistryError::Io` that is neither a
+/// `MemberFault` nor a `WindowFault`, so `run_cell`'s `?` on `append_family`
+/// returns it as a non-containable `ExecFault` (never routed through
+/// `contain`), unlike the `FakeRunner`-planted member faults used elsewhere
+/// in this file.
+fn temp_registry_with_blocked_family_store(name: &str) -> Registry {
+    let dir = std::path::Path::new(env!("CARGO_TARGET_TMPDIR"))
+        .join(format!("aura-campaign-exec-{name}"));
+    let _ = std::fs::remove_dir_all(&dir);
+    std::fs::create_dir_all(&dir).expect("create temp registry dir");
+    std::fs::create_dir_all(dir.join("families.jsonl")).expect("occupy the family store path");
+    Registry::open(dir.join("runs.jsonl"))
+}
+
 fn process(pipeline: Vec<StageBlock>) -> ProcessDoc {
     ProcessDoc {
         format_version: 1,
@@ -955,3 +971,209 @@ fn execute_generalize_keeps_regimes_separate() {
         assert!(g.generalization.is_some(), "two instruments -> graded");
     }
 }
+
+/// C1 across the parallel cell loop: the same campaign under a 1-thread pool
+/// with K=1 and an 8-thread pool with K=2 (fresh registry each) yields an
+/// identical campaign-run record — the campaign-path sibling of the engine's
+/// `*_is_deterministic_across_thread_counts` tests.
+#[test]
+fn campaign_outcome_is_deterministic_across_thread_counts_and_bounds() {
+    let doc = campaign(&["AAA", "BBB", "CCC"]);
+    let proc_doc = process(vec![sweep_stage(false)]);
+    let run_with = |threads: usize, k: usize, name: &str| {
+        let reg = temp_registry(name);
+        let pool = rayon::ThreadPoolBuilder::new()
+            .num_threads(threads)
+            .build()
+            .expect("build pool");
+        pool.install(|| {
+            execute(
+                CAMPAIGN_ID,
+                &doc,
+                &proc_doc,
+                &strategies(),
+                &FakeRunner::clean(),
+                &reg,
+                std::num::NonZeroUsize::new(k).expect("nonzero"),
+            )
+        })
+        .expect("campaign run")
+    };
+    let serial = run_with(1, 1, "det_across_serial");
+    let parallel = run_with(8, 2, "det_across_parallel");
+    assert_eq!(serial.run, parallel.run);
+    assert_eq!(serial.record, parallel.record);
+}
+
+/// #277: a non-containable fault (here, `append_family`'s registry-level IO
+/// refusal, never routed through `contain` since it is neither a
+/// `MemberFault` nor a `WindowFault`) still propagates as a global `Err` out
+/// of the parallel abort-latch loop, and does so with IDENTICAL content
+/// regardless of thread count or pool size. Every cell's sweep hits the SAME
+/// blocked family-store path, so whichever completed cell's `Err` the
+/// doc-order scan in `execute` happens to pick (scheduling-dependent per the
+/// exec.rs doc comment) carries the same IO error kind — exercising the
+/// abort-latch + lowest-doc-order-fault-scan path (and never the
+/// `unreachable!` rebuild arm, since this run returns `Err` before reaching
+/// the rebuild) that the fault-free determinism test above does not.
+#[test]
+fn execute_propagates_a_non_containable_fault_deterministically_under_parallel_chunks() {
+    let doc = campaign(&["AAA", "BBB", "CCC"]);
+    let proc_doc = process(vec![sweep_stage(false)]);
+    let run_with = |threads: usize, k: usize, name: &str| {
+        let reg = temp_registry_with_blocked_family_store(name);
+        let pool = rayon::ThreadPoolBuilder::new()
+            .num_threads(threads)
+            .build()
+            .expect("build pool");
+        pool.install(|| {
+            execute(
+                CAMPAIGN_ID,
+                &doc,
+                &proc_doc,
+                &strategies(),
+                &FakeRunner::clean(),
+                &reg,
+                std::num::NonZeroUsize::new(k).expect("nonzero"),
+            )
+        })
+    };
+    let serial = run_with(1, 1, "fault_prop_serial")
+        .expect_err("a blocked family store is refused, not silently written");
+    let parallel = run_with(8, 3, "fault_prop_parallel")
+        .expect_err("same refusal under real cross-cell parallelism");
+    let kind = |e: &ExecFault| match e {
+        ExecFault::Registry(RegistryError::Io(io)) => io.kind(),
+        other => panic!("expected ExecFault::Registry(Io), got {other:?}"),
+    };
+    assert_eq!(kind(&serial), std::io::ErrorKind::IsADirectory);
+    assert_eq!(kind(&serial), kind(&parallel), "same fault kind regardless of scheduling");
+}
+
+/// Wraps the clean fake and tracks how many DISTINCT instruments are inside
+/// `run_member` at once (refcounted per instrument; high-water mark in
+/// `max_seen`). The 5 ms hold makes overlap near-certain on an 8-thread pool
+/// if the residency bound is absent.
+struct CountingRunner {
+    inner: FakeRunner,
+    live: std::sync::Mutex<BTreeMap<String, usize>>,
+    max_seen: std::sync::atomic::AtomicUsize,
+}
+
+impl MemberRunner for CountingRunner {
+    fn run_member(
+        &self,
+        cell: &CellSpec,
+        params: &[(String, Scalar)],
+        window_ms: (i64, i64),
+    ) -> Result<RunReport, MemberFault> {
+        let distinct = {
+            let mut live = self.live.lock().expect("live instrument map");
+            *live.entry(cell.instrument.clone()).or_insert(0) += 1;
+            live.len()
+        };
+        self.max_seen.fetch_max(distinct, std::sync::atomic::Ordering::SeqCst);
+        std::thread::sleep(std::time::Duration::from_millis(5));
+        let out = self.inner.run_member(cell, params, window_ms);
+        let mut live = self.live.lock().expect("live instrument map");
+        let count = live.get_mut(&cell.instrument).expect("entered above");
+        *count -= 1;
+        if *count == 0 {
+            live.remove(&cell.instrument);
+        }
+        out
+    }
+}
+
+/// The structural residency bound: with K=1 on an 8-thread pool, no two
+/// distinct instruments are ever inside `run_member` at the same time.
+#[test]
+fn parallel_cells_never_exceed_the_instrument_bound() {
+    let reg = temp_registry("instrument_bound");
+    let doc = campaign(&["AAA", "BBB", "CCC"]);
+    let proc_doc = process(vec![sweep_stage(false)]);
+    let runner = CountingRunner {
+        inner: FakeRunner::clean(),
+        live: std::sync::Mutex::new(BTreeMap::new()),
+        max_seen: std::sync::atomic::AtomicUsize::new(0),
+    };
+    let pool = rayon::ThreadPoolBuilder::new().num_threads(8).build().expect("build pool");
+    pool.install(|| {
+        execute(
+            CAMPAIGN_ID,
+            &doc,
+            &proc_doc,
+            &strategies(),
+            &runner,
+            &reg,
+            std::num::NonZeroUsize::new(1).expect("nonzero"),
+        )
+    })
+    .expect("campaign run");
+    let max = runner.max_seen.load(std::sync::atomic::Ordering::SeqCst);
+    assert!(max <= 1, "K=1 must bound distinct live instruments to 1, saw {max}");
+}
+
+/// A registry-write fault is infrastructure, not a cell pathology: it stays
+/// run-fatal in the parallel loop (no cell containment, no partial
+/// campaign-run record).
+#[test]
+fn a_registry_fault_is_run_fatal_in_the_parallel_loop() {
+    let dir = std::path::Path::new(env!("CARGO_TARGET_TMPDIR"))
+        .join("aura-campaign-exec-registry_fatal");
+    let _ = std::fs::remove_dir_all(&dir);
+    std::fs::create_dir_all(&dir).expect("create temp registry dir");
+    // A directory squatting on the family store makes every append_family fail.
+    std::fs::create_dir(dir.join("families.jsonl")).expect("squat the family store");
+    let reg = Registry::open(dir.join("runs.jsonl"));
+    let doc = campaign(&["AAA", "BBB"]);
+    let proc_doc = process(vec![sweep_stage(false)]);
+    let err = execute(
+        CAMPAIGN_ID,
+        &doc,
+        &proc_doc,
+        &strategies(),
+        &FakeRunner::clean(),
+        &reg,
+        DEFAULT_PARALLEL_INSTRUMENTS,
+    )
+    .expect_err("a dead family store is run-fatal");
+    assert!(matches!(err, ExecFault::Registry(_)));
+    assert!(
+        !dir.join("campaign_runs.jsonl").exists(),
+        "no partial campaign-run record on the fatal path"
+    );
+}
+
+/// #272 under the parallel loop: member faults stay contained as failed
+/// cells across concurrently-running cells — the run record persists with
+/// every cell recorded, never a global abort.
+#[test]
+fn member_faults_stay_contained_in_the_parallel_loop() {
+    let reg = temp_registry("parallel_containment");
+    let doc = campaign(&["AAA", "BBB", "CCC"]);
+    let proc_doc = process(vec![sweep_stage(false)]);
+    let point = |fast: i64, slow: i64| {
+        vec![("fast".to_string(), Scalar::i64(fast)), ("slow".to_string(), Scalar::i64(slow))]
+    };
+    let runner = FakeRunner {
+        faults: vec![(point(2, 9), MemberFault::Run("boom".to_string()))],
+    };
+    let pool = rayon::ThreadPoolBuilder::new().num_threads(8).build().expect("build pool");
+    let out = pool
+        .install(|| {
+            execute(
+                CAMPAIGN_ID,
+                &doc,
+                &proc_doc,
+                &strategies(),
+                &runner,
+                &reg,
+                std::num::NonZeroUsize::new(2).expect("nonzero"),
+            )
+        })
+        .expect("member faults are contained, not a global abort");
+    assert_eq!(out.record.cells.len(), 3, "every cell is recorded");
+    assert!(out.record.cells.iter().all(|c| c.fault.is_some()));
+    assert_eq!(reg.load_campaign_runs().expect("load runs").len(), 1);
+}