refactor: cut the engine's backtest-metrics edge via RunReport<M>

C28 phase 2 (Stratification); realizes item 1 of the deferred #147. The
engine's production surface no longer names a backtest-metric type:

- RunReport becomes generic over its metric payload M; sweep/mc/walkforward/
  blueprint thread the parameter (SweepPoint<M>, SweepFamily<M>, WindowRun<M>,
  WalkForwardResult<M>). RunManifest stays concrete and engine-owned (its
  selection: Option<FamilySelection> embeds the foundation-grade analysis type).
- summarize and the MC assembly (McDraw/McFamily/McAggregate/RBootstrap/
  r_bootstrap/monte_carlo) move to aura-backtest - McAggregate::from_draws reads
  RunMetrics fields by name, so generifying it is the phase-6 metric-vocabulary
  abstraction (#147 item 2), still deferred; wholesale relocation is the honest
  cut. The concrete instantiation lives in aura-backtest as
  `type RunReport = aura_engine::RunReport<RunMetrics>` + sibling aliases.
- the statistics kernel (MetricStats/quantile/resample_block/SplitMix64) moves
  to the aura-analysis foundation; the engine re-imports it (inner->foundation,
  legal) and re-exports it so existing consumers stay source-compatible.

Dependency inversion in one commit: aura-engine drops aura-backtest from
[dependencies] (back to dev-deps for its SimBroker/RunMetrics test fixtures);
aura-backtest gains aura-engine. Cycle-free for lib targets - the cycle closes
only through the engine's dev-dep edge, the pattern aura-vocabulary already
uses. aura-backtest reaches the kernel transitively through the engine
re-export, so no aura-backtest -> aura-analysis edge exists (the C28 ladder
permits backtest -> {core, engine} only). run_indexed / SplitMix64::next_f64
widened pub(crate) -> pub for cross-crate use.

Consumers (registry/campaign/cli/composites/ingest/bench) rewired by import
path only, no call-site logic changed. The c28_layering structural test extends
to the full ladder: aura-analysis (no aura-* deps), aura-engine ⊆ {core,
analysis}, aura-backtest ⊆ {core, engine}.

Behaviour-preserving: 1448/0 tests, clippy -D warnings clean, serde shapes
byte-identical (C18 - RunReport<M> keeps field order manifest,metrics; the
CLI pre-serialized-splice contract unchanged), moved code traceable via git
rename detection. Cycle-introduced broken intra-doc links fixed.

closes #292

Cargo.lock

@@ -101,6 +101,7 @@ name = "aura-analysis"
 version = "0.1.0"
 dependencies = [
  "serde",
+ "serde_json",
 ]
 
 [[package]]
@@ -108,6 +109,10 @@ name = "aura-backtest"
 version = "0.1.0"
 dependencies = [
  "aura-core",
+ "aura-engine",
+ "aura-std",
+ "aura-strategy",
+ "rayon",
  "serde",
  "serde_json",
 ]
@@ -117,6 +122,7 @@ name = "aura-bench"
 version = "0.1.0"
 dependencies = [
  "aura-analysis",
+ "aura-backtest",
  "aura-core",
  "aura-engine",
  "aura-ingest",
@@ -236,6 +242,8 @@ dependencies = [
 name = "aura-registry"
 version = "0.1.0"
 dependencies = [
+ "aura-analysis",
+ "aura-backtest",
  "aura-core",
  "aura-engine",
  "aura-research",

crates/aura-analysis/Cargo.toml

@@ -11,3 +11,8 @@ publish.workspace = true
 # (de)serialization path for the run registry (C18). Deterministic output
 # (C1-safe). The backtest metric types moved to aura-backtest (#291).
 serde = { workspace = true }
+
+[dev-dependencies]
+# serde_json is used only by the moved MetricStats round-trip test (#291 Task
+# 1); production JSON rendering lives at the engine/backtest boundary.
+serde_json = { workspace = true }

crates/aura-analysis/src/lib.rs

@@ -102,6 +102,102 @@ pub fn expected_max_of_normals(k: usize) -> f64 {
         + GAMMA * inv_norm_cdf(1.0 - 1.0 / (kf * std::f64::consts::E))
 }
 
+/// A tiny, fully-deterministic, dependency-free PRNG (SplitMix64). The seed
+/// completely determines the sequence; no external entropy, no global state.
+/// Bit-stable across toolchains and crate versions — the property C1 needs for
+/// seed-as-input reproducibility (C12).
+pub struct SplitMix64 {
+    state: u64,
+}
+
+impl SplitMix64 {
+    pub fn new(seed: u64) -> Self {
+        Self { state: seed }
+    }
+    pub fn next_u64(&mut self) -> u64 {
+        self.state = self.state.wrapping_add(0x9E37_79B9_7F4A_7C15);
+        let mut z = self.state;
+        z = (z ^ (z >> 30)).wrapping_mul(0xBF58_476D_1CE4_E5B9);
+        z = (z ^ (z >> 27)).wrapping_mul(0x94D0_49BB_1331_11EB);
+        z ^ (z >> 31)
+    }
+    /// A uniform `f64` in `[0, 1)` from the top 53 bits.
+    pub fn next_f64(&mut self) -> f64 {
+        (self.next_u64() >> 11) as f64 / ((1u64 << 53) as f64)
+    }
+}
+
+/// Mean + a fixed quantile set of one metric across the realizations. `p50` is
+/// the median (the two coincide by definition), so "mean/median/quantiles" is
+/// `mean` + `p50` + the surrounding quantiles, with no redundant field.
+#[derive(Clone, Debug, PartialEq, serde::Serialize, serde::Deserialize)]
+pub struct MetricStats {
+    pub mean: f64,
+    pub p5: f64,
+    pub p25: f64,
+    pub p50: f64, // == median
+    pub p75: f64,
+    pub p95: f64,
+}
+
+impl MetricStats {
+    /// Mean + the fixed type-7 quantile set over a value set. Sorts a copy;
+    /// `values` must be finite and non-empty. The shared reduction behind both the
+    /// MC aggregate (a metric across draws) and the walk-forward param-stability
+    /// summary (a param across windows; `aura-engine`'s `param_stability`).
+    pub fn from_values(values: &[f64]) -> MetricStats {
+        let mut xs = values.to_vec();
+        xs.sort_by(|a, b| a.partial_cmp(b).expect("values are finite"));
+        MetricStats {
+            mean: xs.iter().sum::<f64>() / xs.len() as f64,
+            p5: quantile(&xs, 0.05),
+            p25: quantile(&xs, 0.25),
+            p50: quantile(&xs, 0.50),
+            p75: quantile(&xs, 0.75),
+            p95: quantile(&xs, 0.95),
+        }
+    }
+}
+
+/// Linear-interpolation quantile (the numpy/Excel "type 7" default) over a
+/// pre-sorted, non-empty slice. `p` in `[0, 1]`. `rank = p * (n - 1)`;
+/// interpolate between the two bracketing order statistics. `n == 1` returns the
+/// sole value.
+pub fn quantile(sorted: &[f64], p: f64) -> f64 {
+    let n = sorted.len();
+    if n == 1 {
+        return sorted[0];
+    }
+    let rank = p * (n - 1) as f64;
+    let lo = rank.floor() as usize;
+    let frac = rank - lo as f64;
+    if lo + 1 < n {
+        sorted[lo] + frac * (sorted[lo + 1] - sorted[lo])
+    } else {
+        sorted[n - 1]
+    }
+}
+
+/// One moving-block resample of `rs` to length `rs.len()` (non-circular; the final
+/// block is truncated so the resample has exactly `n` values). `block_len` must be
+/// pre-clamped to `[1, rs.len()]` by the caller. Pure given the `rng` state — the
+/// shared kernel behind `r_bootstrap` and the trials-deflation reality-check (C1).
+pub fn resample_block(rs: &[f64], block_len: usize, rng: &mut SplitMix64) -> Vec<f64> {
+    let n = rs.len();
+    debug_assert!(
+        (1..=n).contains(&block_len),
+        "resample_block requires block_len in [1, rs.len()] (caller pre-clamps); \
+         got block_len={block_len}, rs.len()={n}",
+    );
+    let mut sample: Vec<f64> = Vec::with_capacity(n);
+    while sample.len() < n {
+        let start = (rng.next_u64() % (n - block_len + 1) as u64) as usize;
+        let take = block_len.min(n - sample.len());
+        sample.extend_from_slice(&rs[start..start + take]);
+    }
+    sample
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -125,4 +221,73 @@ mod tests {
         assert!((1.0..1.1).contains(&e4));           // ~1.05 (not the √(2 ln 4)=1.66 asymptote)
         assert!((2.4..2.7).contains(&e100));         // ~2.53
     }
+
+    #[test]
+    fn quantile_endpoints_and_singleton() {
+        // singleton returns the sole value; p=0 -> min, p=1 -> max.
+        assert_eq!(quantile(&[42.0], 0.0), 42.0);
+        assert_eq!(quantile(&[42.0], 0.5), 42.0);
+        assert_eq!(quantile(&[42.0], 1.0), 42.0);
+        let xs = [1.0, 2.0, 3.0, 4.0, 5.0];
+        assert_eq!(quantile(&xs, 0.0), 1.0);
+        assert_eq!(quantile(&xs, 1.0), 5.0);
+    }
+
+    #[test]
+    fn metric_stats_from_values_matches_known_fixture() {
+        // type-7 quantile + mean over [0,1,2,3,4], directly on the
+        // extracted reduction: mean=2.0, p50=2.0, p5≈0.2, p95≈3.8 (same numbers the
+        // aggregate fixture pins, now on from_values).
+        let s = MetricStats::from_values(&[0.0, 1.0, 2.0, 3.0, 4.0]);
+        assert_eq!(s.mean, 2.0);
+        assert_eq!(s.p50, 2.0);
+        assert!((s.p5 - 0.2).abs() < 1e-9, "p5 = {}", s.p5);
+        assert!((s.p95 - 3.8).abs() < 1e-9, "p95 = {}", s.p95);
+    }
+
+    #[test]
+    fn metric_stats_serde_round_trips() {
+        // MetricStats gains serde (consistent with the
+        // report types) so the CLI summary renders it and #70 lineage can persist.
+        let s = MetricStats::from_values(&[1.0, 2.0, 3.0]);
+        let json = serde_json::to_string(&s).expect("serialize MetricStats");
+        let back: MetricStats = serde_json::from_str(&json).expect("deserialize MetricStats");
+        assert_eq!(back, s);
+    }
+
+    #[test]
+    fn resample_block_returns_n_values_from_contiguous_runs() {
+        // The extracted kernel's standalone contract (now a public entry point):
+        // given a pre-clamped block_len in [1, n], it yields exactly `n` values, each
+        // appearing in `rs`, arranged as contiguous runs of `rs` (the final run
+        // truncated to reach exactly n). Distinct powers of ten make membership a sum
+        // check. n=5, block_len=2 -> runs of length 2, 2, 1.
+        let rs = [1.0, 10.0, 100.0, 1000.0, 10000.0];
+        let n = rs.len();
+        let block_len = 2;
+        let mut rng = SplitMix64::new(7);
+        let sample = resample_block(&rs, block_len, &mut rng);
+        assert_eq!(sample.len(), n, "resample has exactly n values");
+        assert!(
+            sample.iter().all(|v| rs.contains(v)),
+            "every resampled value is drawn from rs",
+        );
+        // Reconstruct the contiguous-run structure: walk the sample in blocks of
+        // `block_len` (last truncated) and assert each block is a contiguous slice of rs.
+        let mut filled = 0usize;
+        while filled < n {
+            let take = block_len.min(n - filled);
+            let run = &sample[filled..filled + take];
+            let start = rs
+                .iter()
+                .position(|v| v == &run[0])
+                .expect("run head is in rs");
+            assert_eq!(
+                run,
+                &rs[start..start + take],
+                "each run is a contiguous slice of rs",
+            );
+            filled += take;
+        }
+    }
 }

crates/aura-backtest/Cargo.toml

@@ -7,6 +7,14 @@ publish.workspace = true
 
 [dependencies]
 aura-core = { path = "../aura-core" }
+# aura-engine is the metric-generic engine this crate instantiates with
+# `M = RunMetrics` (`RunReport`/`SweepFamily`/etc. aliases), whose
+# `run_indexed` the moved MC assembly drives, and whose re-exported
+# `MetricStats`/`resample_block`/`SplitMix64` (foundation-grade, from
+# aura-analysis) the moved `mc.rs` reduces through — no direct aura-analysis
+# edge, keeping the C28 ladder one-hop (#292, C28 phase 2). Outer-rung
+# dependency — the engine itself only dev-depends back on this crate.
+aura-engine = { path = "../aura-engine" }
 # serde gives the backtest metric types (`RunMetrics`/`RMetrics`/`PositionEvent`)
 # their typed (de)serialization path for the run registry (C18; moved with the
 # types from aura-analysis, #291). Deterministic output (C1-safe).
@@ -17,3 +25,13 @@ serde = { workspace = true }
 # for production lives in aura-engine's RunReport::to_json). Test-only, so it
 # never enters the frozen artifact's library link (C16).
 serde_json = { workspace = true }
+# aura-std / aura-strategy back the moved `mc.rs`/`metrics.rs` test fixtures
+# (Sma/Sub/Recorder, Bias) — the crate-private `aura-engine::test_fixtures`
+# harness is unreachable from here, so the fixture is reconstructed through
+# the public node vocabulary instead (test-only, mirrors the aura-engine
+# integration tests' own pattern).
+aura-std = { path = "../aura-std" }
+aura-strategy = { path = "../aura-strategy" }
+# rayon backs the thread-count-explicit `monte_carlo_with_threads` test
+# wrapper (moved verbatim from aura-engine's mc.rs, #291) — test-only.
+rayon = "1"

crates/aura-backtest/src/lib.rs

@@ -1,15 +1,22 @@
 //! `aura-backtest` — the backtest layer (C28): simulated execution without
 //! money (the `SimBroker` pip yardstick, position management), measured in R,
 //! plus the backtest metric reductions over the recorded streams
-//! (`metrics`, moved from `aura-analysis` by #291). Depends only on
+//! (`metrics`, moved from `aura-analysis` by #291) and the Monte-Carlo
-//! `aura-core` (+ serde for the metric types' C18 wire shapes).
+//! assembly (`mc`, moved from `aura-engine` by #291/#292, C28 phase 2). The
+//! outer rung of the metric-generic engine (C28): `RunReport` here is the
+//! concrete `M = RunMetrics` instantiation of `aura_engine::RunReport<M>`.
 
+mod mc;
 mod metrics;
 mod position_management;
 mod sim_broker;
 
+// `assemble_mc` stays module-private (it was never `pub` in the pre-#291 engine
+// lib.rs re-export either — only its two callers, `monte_carlo` and the
+// `#[cfg(test)]` thread-count wrapper, are public/test surface).
+pub use mc::{monte_carlo, r_bootstrap, McAggregate, McDraw, McFamily, RBootstrap};
 pub use metrics::{
-    derive_position_events, r_metrics_from_rs, summarize_r, PositionAction,
+    derive_position_events, r_metrics_from_rs, summarize, summarize_r, PositionAction,
     PositionEvent, RMetrics, RunMetrics,
 };
 pub use position_management::{
@@ -17,3 +24,12 @@ pub use position_management::{
     RECORD_KINDS as PM_RECORD_KINDS, WIDTH as PM_WIDTH,
 };
 pub use sim_broker::SimBroker;
+
+/// The trading instantiation of the engine's metric-generic run record (C28
+/// phase 2, #292): source-compatible concrete names for every consumer that
+/// deals in backtest metrics.
+pub type RunReport = aura_engine::RunReport<RunMetrics>;
+pub type SweepPoint = aura_engine::SweepPoint<RunMetrics>;
+pub type SweepFamily = aura_engine::SweepFamily<RunMetrics>;
+pub type WindowRun = aura_engine::WindowRun<RunMetrics>;
+pub type WalkForwardResult = aura_engine::WalkForwardResult<RunMetrics>;

crates/aura-backtest/src/mc.rs

@@ -3,18 +3,25 @@
 //! a seed set — each seed a disjoint C1 realization — and collects an
 //! [`McFamily`]: the per-seed [`McDraw`]s in seed-input order plus a stored
 //! [`McAggregate`] (mean + quantiles of all three run metrics across the
-//! realizations). It reuses the disjoint-parallel [`run_indexed`](crate::sweep)
+//! realizations). It reuses the disjoint-parallel [`run_indexed`](aura_engine::run_indexed)
 //! core `sweep` drives — the varying dimension is the *seed*, not a tuning param.
 //! Eager-agnostic (C12/#71): the API takes seeds + a per-draw closure, never a
 //! materialized stream `Vec`; the seed -> `Source` construction is a closure-body
-//! concern.
+//! concern. Moved verbatim from `aura-engine::mc` (#291, C28 phase 2): the
+//! engine stays metric-agnostic, so the concrete `RunMetrics` instantiation of
+//! `RunReport` lives here, the outer C28 rung.
 
-use crate::sweep::run_indexed;
+// MetricStats/resample_block/SplitMix64 are foundation-grade (aura-analysis),
-use crate::{RunMetrics, RunReport, Scalar};
+// but reached here through aura-engine's re-export rather than a direct
-use crate::harness::SplitMix64;
+// aura-analysis dependency — the C28 ladder (c28_layering) permits
+// aura-backtest -> {aura-core, aura-engine} only, not a same-rung-skipping
+// direct edge to the foundation crate.
+use aura_core::Scalar;
+use aura_engine::{run_indexed, resample_block, MetricStats, SplitMix64};
+use crate::{RunMetrics, RunReport};
 
 /// One Monte-Carlo realization: the seed that drove it and the full `RunReport`.
-/// Self-describing, analog to [`SweepPoint`](crate::SweepPoint).
+/// Self-describing, analog to [`SweepPoint`](aura_engine::SweepPoint).
 #[derive(Clone, Debug, PartialEq)]
 pub struct McDraw {
     pub seed: u64,
@@ -23,7 +30,7 @@ pub struct McDraw {
 
 /// The result family of a Monte-Carlo run — one [`McDraw`] per seed, in
 /// seed-**input** order (independent of thread completion), plus the stored
-/// [`McAggregate`]. Analog to [`SweepFamily`](crate::SweepFamily).
+/// [`McAggregate`]. Analog to [`SweepFamily`](aura_engine::SweepFamily).
 #[derive(Clone, Debug, PartialEq)]
 pub struct McFamily {
     pub draws: Vec<McDraw>,
@@ -41,38 +48,6 @@ pub struct McAggregate {
     pub bias_sign_flips: MetricStats,
 }
 
-/// Mean + a fixed quantile set of one metric across the realizations. `p50` is
-/// the median (the two coincide by definition), so "mean/median/quantiles" is
-/// `mean` + `p50` + the surrounding quantiles, with no redundant field.
-#[derive(Clone, Debug, PartialEq, serde::Serialize, serde::Deserialize)]
-pub struct MetricStats {
-    pub mean: f64,
-    pub p5: f64,
-    pub p25: f64,
-    pub p50: f64, // == median
-    pub p75: f64,
-    pub p95: f64,
-}
-
-impl MetricStats {
-    /// Mean + the fixed type-7 quantile set over a value set. Sorts a copy;
-    /// `values` must be finite and non-empty. The shared reduction behind both the
-    /// MC aggregate (a metric across draws) and the walk-forward param-stability
-    /// summary (a param across windows, [`crate::param_stability`]).
-    pub fn from_values(values: &[f64]) -> MetricStats {
-        let mut xs = values.to_vec();
-        xs.sort_by(|a, b| a.partial_cmp(b).expect("values are finite"));
-        MetricStats {
-            mean: xs.iter().sum::<f64>() / xs.len() as f64,
-            p5: quantile(&xs, 0.05),
-            p25: quantile(&xs, 0.25),
-            p50: quantile(&xs, 0.50),
-            p75: quantile(&xs, 0.75),
-            p95: quantile(&xs, 0.95),
-        }
-    }
-}
-
 impl McAggregate {
     /// Pure reduction over the realizations — recomputable from `draws` alone (it
     /// is exactly what [`monte_carlo`] stored). `draws` must be non-empty (a
@@ -91,25 +66,6 @@ impl McAggregate {
     }
 }
 
-/// Linear-interpolation quantile (the numpy/Excel "type 7" default) over a
-/// pre-sorted, non-empty slice. `p` in `[0, 1]`. `rank = p * (n - 1)`;
-/// interpolate between the two bracketing order statistics. `n == 1` returns the
-/// sole value.
-fn quantile(sorted: &[f64], p: f64) -> f64 {
-    let n = sorted.len();
-    if n == 1 {
-        return sorted[0];
-    }
-    let rank = p * (n - 1) as f64;
-    let lo = rank.floor() as usize;
-    let frac = rank - lo as f64;
-    if lo + 1 < n {
-        sorted[lo] + frac * (sorted[lo + 1] - sorted[lo])
-    } else {
-        sorted[n - 1]
-    }
-}
-
 /// The single assembly path behind both [`monte_carlo`] and the thread-count-
 /// explicit test wrapper: run `run_one` over every seed via `run_indexed` (on
 /// whichever pool is ambient when called), build the per-seed draws in
@@ -139,7 +95,7 @@ where
 ///
 /// Precondition: `seeds` is non-empty (a Monte-Carlo over zero realizations has
 /// no defined aggregate). A `debug_assert!` guards it; the `-> McFamily`
-/// signature is preserved (no `Result`), matching [`sweep`](crate::sweep).
+/// signature is preserved (no `Result`), matching [`sweep`](aura_engine::sweep).
 pub fn monte_carlo<F>(base_point: &[Scalar], seeds: &[u64], run_one: F) -> McFamily
 where
     F: Fn(u64, &[Scalar]) -> RunReport + Sync,
@@ -186,26 +142,6 @@ pub struct RBootstrap {
     pub n_resamples: usize,
 }
 
-/// One moving-block resample of `rs` to length `rs.len()` (non-circular; the final
-/// block is truncated so the resample has exactly `n` values). `block_len` must be
-/// pre-clamped to `[1, rs.len()]` by the caller. Pure given the `rng` state — the
-/// shared kernel behind `r_bootstrap` and the trials-deflation reality-check (C1).
-pub fn resample_block(rs: &[f64], block_len: usize, rng: &mut SplitMix64) -> Vec<f64> {
-    let n = rs.len();
-    debug_assert!(
-        (1..=n).contains(&block_len),
-        "resample_block requires block_len in [1, rs.len()] (caller pre-clamps); \
-         got block_len={block_len}, rs.len()={n}",
-    );
-    let mut sample: Vec<f64> = Vec::with_capacity(n);
-    while sample.len() < n {
-        let start = (rng.next_u64() % (n - block_len + 1) as u64) as usize;
-        let take = block_len.min(n - sample.len());
-        sample.extend_from_slice(&rs[start..start + take]);
-    }
-    sample
-}
-
 /// Moving-block bootstrap of `rs` (non-circular; the final block of each resample is
 /// truncated so the resample has exactly `n` values). `block_len` is clamped to
 /// `[1, n]`. Empty `rs` or zero resamples -> an all-zero `RBootstrap`. Pure given
@@ -236,9 +172,83 @@ pub fn r_bootstrap(rs: &[f64], n_resamples: usize, block_len: usize, seed: u64)
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::test_fixtures::composite_sma_cross_harness;
+    use aura_engine::{
-    use crate::{f64_field, summarize, RunManifest, SyntheticSpec, Timestamp};
+        f64_field, BlueprintNode, Composite, Edge, OutField, Role, RunManifest, SyntheticSpec,
-    use aura_core::Scalar;
+        Target, Timestamp,
+    };
+    use aura_core::{Firing, Scalar, ScalarKind};
+    use aura_std::{Recorder, Sma, Sub};
+    use aura_strategy::Bias;
+    use crate::{summarize, SimBroker};
+    use std::sync::mpsc;
+
+    // The SMA-cross signal-quality harness fixture, reconstructed here through
+    // the PUBLIC aura-engine/aura-std/aura-strategy/aura-backtest surface —
+    // `aura-engine::test_fixtures` is `#[cfg(test)]`-private to the engine crate
+    // itself and unreachable from here (the same reason `random_sweep_e2e.rs` /
+    // `list_sweep_e2e.rs` carry their own copy). Bodies verbatim from
+    // `aura-engine/src/test_fixtures.rs`'s `sma_cross`/`composite_sma_cross_harness`.
+
+    /// The SMA-cross signal as a reusable value-empty composite: one input role
+    /// (price), one output (fast-minus-slow spread). Lengths injected at compile.
+    fn sma_cross() -> Composite {
+        Composite::new(
+            "sma_cross",
+            vec![
+                Sma::builder().named("fast").into(),
+                Sma::builder().named("slow").into(),
+                Sub::builder().into(),
+            ],
+            vec![
+                Edge { from: 0, to: 2, slot: 0, from_field: 0 },
+                Edge { from: 1, to: 2, slot: 1, from_field: 0 },
+            ],
+            vec![Role {
+                name: "price".into(),
+                targets: vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }],
+                source: None,
+            }],
+            vec![OutField { node: 2, field: 0, name: "out".into() }],
+        )
+    }
+
+    /// The signal-quality harness as a value-empty composite blueprint with two
+    /// recording sinks (equity, exposure).
+    #[allow(clippy::type_complexity)]
+    fn composite_sma_cross_harness() -> (
+        Composite,
+        mpsc::Receiver<(Timestamp, Vec<Scalar>)>,
+        mpsc::Receiver<(Timestamp, Vec<Scalar>)>,
+    ) {
+        let (tx_eq, rx_eq) = mpsc::channel();
+        let (tx_ex, rx_ex) = mpsc::channel();
+        let bp = Composite::new(
+            "root",
+            vec![
+                BlueprintNode::Composite(sma_cross()),
+                Bias::builder().named("bias").into(),
+                SimBroker::builder(0.0001).into(),
+                Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx_eq).into(),
+                Recorder::builder(vec![ScalarKind::F64], Firing::Any, tx_ex).into(),
+            ],
+            vec![
+                Edge { from: 0, to: 1, slot: 0, from_field: 0 }, // composite out -> Bias
+                Edge { from: 1, to: 2, slot: 0, from_field: 0 }, // exposure -> broker slot 0
+                Edge { from: 2, to: 3, slot: 0, from_field: 0 }, // equity -> sink
+                Edge { from: 1, to: 4, slot: 0, from_field: 0 }, // exposure -> sink
+            ],
+            vec![Role {
+                name: "src".into(),
+                targets: vec![
+                    Target { node: 0, slot: 0 }, // price -> sma_cross role 0
+                    Target { node: 2, slot: 1 }, // price -> SimBroker price slot
+                ],
+                source: Some(ScalarKind::F64),
+            }],
+            vec![], // output: the root ends in sinks
+        );
+        (bp, rx_eq, rx_ex)
+    }
 
     /// Build + bootstrap + run + drain + summarize one (seed, point) into a
     /// `RunReport`. A free `fn` (Copy + Sync) so it serves as the `monte_carlo`
@@ -371,75 +381,6 @@ mod tests {
         assert!((agg.total_pips.p95 - 3.8).abs() < 1e-9, "p95 = {}", agg.total_pips.p95);
     }
 
-    #[test]
-    fn quantile_endpoints_and_singleton() {
-        // singleton returns the sole value; p=0 -> min, p=1 -> max.
-        assert_eq!(quantile(&[42.0], 0.0), 42.0);
-        assert_eq!(quantile(&[42.0], 0.5), 42.0);
-        assert_eq!(quantile(&[42.0], 1.0), 42.0);
-        let xs = [1.0, 2.0, 3.0, 4.0, 5.0];
-        assert_eq!(quantile(&xs, 0.0), 1.0);
-        assert_eq!(quantile(&xs, 1.0), 5.0);
-    }
-
-    #[test]
-    fn metric_stats_from_values_matches_known_fixture() {
-        // type-7 quantile + mean over [0,1,2,3,4], directly on the
-        // extracted reduction: mean=2.0, p50=2.0, p5≈0.2, p95≈3.8 (same numbers the
-        // aggregate fixture pins, now on from_values).
-        let s = MetricStats::from_values(&[0.0, 1.0, 2.0, 3.0, 4.0]);
-        assert_eq!(s.mean, 2.0);
-        assert_eq!(s.p50, 2.0);
-        assert!((s.p5 - 0.2).abs() < 1e-9, "p5 = {}", s.p5);
-        assert!((s.p95 - 3.8).abs() < 1e-9, "p95 = {}", s.p95);
-    }
-
-    #[test]
-    fn metric_stats_serde_round_trips() {
-        // MetricStats gains serde (consistent with the
-        // report types) so the CLI summary renders it and #70 lineage can persist.
-        let s = MetricStats::from_values(&[1.0, 2.0, 3.0]);
-        let json = serde_json::to_string(&s).expect("serialize MetricStats");
-        let back: MetricStats = serde_json::from_str(&json).expect("deserialize MetricStats");
-        assert_eq!(back, s);
-    }
-
-    #[test]
-    fn resample_block_returns_n_values_from_contiguous_runs() {
-        // The extracted kernel's standalone contract (now a public entry point):
-        // given a pre-clamped block_len in [1, n], it yields exactly `n` values, each
-        // appearing in `rs`, arranged as contiguous runs of `rs` (the final run
-        // truncated to reach exactly n). Distinct powers of ten make membership a sum
-        // check. n=5, block_len=2 -> runs of length 2, 2, 1.
-        let rs = [1.0, 10.0, 100.0, 1000.0, 10000.0];
-        let n = rs.len();
-        let block_len = 2;
-        let mut rng = SplitMix64::new(7);
-        let sample = resample_block(&rs, block_len, &mut rng);
-        assert_eq!(sample.len(), n, "resample has exactly n values");
-        assert!(
-            sample.iter().all(|v| rs.contains(v)),
-            "every resampled value is drawn from rs",
-        );
-        // Reconstruct the contiguous-run structure: walk the sample in blocks of
-        // `block_len` (last truncated) and assert each block is a contiguous slice of rs.
-        let mut filled = 0usize;
-        while filled < n {
-            let take = block_len.min(n - filled);
-            let run = &sample[filled..filled + take];
-            let start = rs
-                .iter()
-                .position(|v| v == &run[0])
-                .expect("run head is in rs");
-            assert_eq!(
-                run,
-                &rs[start..start + take],
-                "each run is a contiguous slice of rs",
-            );
-            filled += take;
-        }
-    }
-
     #[test]
     fn r_bootstrap_empty_series_is_all_zero() {
         let b = r_bootstrap(&[], 100, 1, 7);

crates/aura-backtest/src/metrics.rs

@@ -8,7 +8,34 @@
 //! Every type's serde shape is byte-pinned by C18 goldens; no float expression
 //! re-associated.
 
-use aura_core::{Scalar, Timestamp};
+use aura_core::{Scalar, SeriesFold, Timestamp};
+
+/// Reduce a run's recorded pip-equity + exposure streams into summary metrics.
+/// Pure — identical inputs yield identical metrics (C1/C12). Timestamps are
+/// carried in the input to match exactly what a sink records; the reduction
+/// itself is value-only (it does not read the timestamps).
+pub fn summarize(
+    equity: &[(Timestamp, f64)],
+    exposure: &[(Timestamp, f64)],
+) -> RunMetrics {
+    // total_pips + max_drawdown fold over equity; sign_flips folds over exposure.
+    // SeriesFold is the single arithmetic source of truth (shared with the
+    // SeriesReducer sink), so this is byte-identical to the prior inline loops.
+    let mut eqf = SeriesFold::new();
+    for &(_, v) in equity {
+        eqf.fold(v);
+    }
+    let mut exf = SeriesFold::new();
+    for &(_, v) in exposure {
+        exf.fold(v);
+    }
+    RunMetrics {
+        total_pips: eqf.last,
+        max_drawdown: eqf.max_drawdown,
+        bias_sign_flips: exf.sign_flips,
+        r: None,
+    }
+}
 
 /// Summary metrics reduced from a run's recorded streams — the `-> metrics`
 /// half of C12's atomic sim unit. Pure function of the recorded streams.
@@ -1048,4 +1075,181 @@ mod tests {
         assert_eq!(m.expectancy_r, 0.0);
         assert_eq!(m.sqn, 0.0);
     }
+
+    // The `summarize` end-to-end + unit tests, moved verbatim from
+    // `aura-engine::report` (#291, C28 phase 2): `run_once` drives a real
+    // bootstrapped Harness through public aura-engine/aura-std/aura-strategy
+    // APIs, so it needs no engine-private item.
+    use aura_core::{Firing, NodeSchema, PortSpec, ScalarKind};
+    use aura_engine::{f64_field, Edge, FlatGraph, Harness, RunManifest, RunReport, SourceSpec, Target, VecSource};
+    use aura_std::{Recorder, Sma, Sub};
+    use aura_strategy::Bias;
+    use crate::SimBroker;
+    use std::sync::mpsc;
+
+    /// The declared signature of a `Recorder` over one f64 column (the sink shape
+    /// the two-sink harness uses).
+    fn f64_recorder_sig() -> NodeSchema {
+        NodeSchema {
+            inputs: vec![PortSpec { kind: ScalarKind::F64, firing: Firing::Any, name: "in".into() }],
+            output: vec![],
+            params: vec![],
+        }
+    }
+
+    /// Build an f64 source stream from (timestamp, value) points (mirrors the
+    /// harness.rs test helper; the e2e test needs its own copy — the harness
+    /// test module's is private to that module).
+    fn f64_stream(points: &[(i64, f64)]) -> Vec<(Timestamp, Scalar)> {
+        points.iter().map(|&(t, v)| (Timestamp(t), Scalar::f64(v))).collect()
+    }
+
+    /// Bootstrap the cycle-0007 signal-quality harness with TWO sinks: one on
+    /// the SimBroker equity output (node 4 -> node 5) and one on the Bias
+    /// output (node 3 -> node 6). Returns the harness plus the two receivers.
+    #[allow(clippy::type_complexity)]
+    fn build_two_sink_harness() -> (
+        Harness,
+        mpsc::Receiver<(Timestamp, Vec<Scalar>)>,
+        mpsc::Receiver<(Timestamp, Vec<Scalar>)>,
+    ) {
+        let (tx_eq, rx_eq) = mpsc::channel();
+        let (tx_ex, rx_ex) = mpsc::channel();
+        let h = Harness::bootstrap(FlatGraph {
+            nodes: vec![
+                Box::new(Sma::new(2)),                                       // 0
+                Box::new(Sma::new(4)),                                       // 1
+                Box::new(Sub::new()),                                        // 2
+                Box::new(Bias::new(0.5)),                                    // 3
+                Box::new(SimBroker::new(0.0001)),                            // 4
+                Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx_eq)), // 5 equity sink
+                Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx_ex)), // 6 exposure sink
+            ],
+            signatures: vec![
+                Sma::builder().schema().clone(),
+                Sma::builder().schema().clone(),
+                Sub::builder().schema().clone(),
+                Bias::builder().schema().clone(),
+                SimBroker::builder(0.0001).schema().clone(),
+                f64_recorder_sig(),
+                f64_recorder_sig(),
+            ],
+            sources: vec![SourceSpec::raw(ScalarKind::F64, vec![
+                    Target { node: 0, slot: 0 },
+                    Target { node: 1, slot: 0 },
+                    Target { node: 4, slot: 1 }, // price into the broker
+                ])],
+            edges: vec![
+                Edge { from: 0, to: 2, slot: 0, from_field: 0 },
+                Edge { from: 1, to: 2, slot: 1, from_field: 0 },
+                Edge { from: 2, to: 3, slot: 0, from_field: 0 },
+                Edge { from: 3, to: 4, slot: 0, from_field: 0 },
+                Edge { from: 4, to: 5, slot: 0, from_field: 0 }, // equity -> sink 5
+                Edge { from: 3, to: 6, slot: 0, from_field: 0 }, // exposure -> sink 6
+            ],
+            taps: Vec::new(),
+        })
+        .expect("valid signal-quality DAG");
+        (h, rx_eq, rx_ex)
+    }
+
+    fn run_once() -> RunReport<RunMetrics> {
+        let (mut h, rx_eq, rx_ex) = build_two_sink_harness();
+        h.run(vec![Box::new(VecSource::new(f64_stream(&[
+            (1, 1.0000),
+            (2, 1.0010),
+            (3, 1.0025),
+            (4, 1.0020),
+            (5, 1.0040),
+        ])))]);
+        let eq_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_eq.try_iter().collect();
+        let ex_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_ex.try_iter().collect();
+        let equity = f64_field(&eq_rows, 0);
+        let exposure = f64_field(&ex_rows, 0);
+        let metrics = summarize(&equity, &exposure);
+        RunReport {
+            manifest: RunManifest {
+                commit: "test-commit".to_string(),
+                params: vec![
+                    ("sma_fast".to_string(), Scalar::i64(2)),
+                    ("sma_slow".to_string(), Scalar::i64(4)),
+                    ("bias_scale".to_string(), Scalar::f64(0.5)),
+                ],
+                defaults: vec![],
+                window: (Timestamp(1), Timestamp(5)),
+                seed: 0,
+                broker: "sim-optimal(pip_size=0.0001)".to_string(),
+                selection: None,
+                instrument: None,
+                topology_hash: None,
+                project: None,
+            },
+            metrics,
+        }
+    }
+
+    #[test]
+    fn report_is_deterministic_end_to_end() {
+        let r1 = run_once();
+        let r2 = run_once();
+        // a run actually emitted metrics over a non-empty pip curve
+        assert!(r1.metrics.total_pips.is_finite());
+        // same manifest -> same metrics (C1/C12): two runs are bit-identical
+        assert_eq!(r1.metrics, r2.metrics);
+        assert_eq!(r1.to_json(), r2.to_json());
+    }
+
+    fn samples(values: &[f64]) -> Vec<(Timestamp, f64)> {
+        values
+            .iter()
+            .enumerate()
+            .map(|(i, &v)| (Timestamp(i as i64 + 1), v))
+            .collect()
+    }
+
+    #[test]
+    fn summarize_total_pips_is_last_cumulative_value() {
+        let equity = samples(&[0.0, 5.0, 4.0, 12.0]);
+        let m = summarize(&equity, &[]);
+        assert_eq!(m.total_pips, 12.0);
+    }
+
+    #[test]
+    fn summarize_is_zero_on_empty_streams() {
+        let m = summarize(&[], &[]);
+        assert_eq!(m.total_pips, 0.0);
+        assert_eq!(m.max_drawdown, 0.0);
+        assert_eq!(m.bias_sign_flips, 0);
+    }
+
+    #[test]
+    fn summarize_max_drawdown_is_worst_peak_to_trough() {
+        // peak 10 then trough 5 (drop 5), recovers to 8; worst drop is 5,
+        // not the final drop (10 -> 8 = 2).
+        let equity = samples(&[0.0, 10.0, 5.0, 8.0]);
+        let m = summarize(&equity, &[]);
+        assert_eq!(m.max_drawdown, 5.0);
+    }
+
+    #[test]
+    fn summarize_max_drawdown_zero_on_monotonic_curve() {
+        let equity = samples(&[0.0, 1.0, 2.0, 3.0]);
+        let m = summarize(&equity, &[]);
+        assert_eq!(m.max_drawdown, 0.0);
+    }
+
+    #[test]
+    fn summarize_sign_flips_counts_signum_changes() {
+        // signum series: + + - 0 -  ->  flips at +->-, -->0, 0->- = 3.
+        let exposure = samples(&[0.5, 0.5, -0.5, 0.0, -0.5]);
+        let m = summarize(&[], &exposure);
+        assert_eq!(m.bias_sign_flips, 3);
+    }
+
+    #[test]
+    fn summarize_sign_flips_zero_on_constant_sign() {
+        let exposure = samples(&[0.2, 0.5, 1.0, 0.7]);
+        let m = summarize(&[], &exposure);
+        assert_eq!(m.bias_sign_flips, 0);
+    }
 }

crates/aura-bench/Cargo.toml

@@ -36,3 +36,7 @@ zip = "2"
 # fingerprint coverage.
 aura-registry = { path = "../aura-registry" }
 aura-analysis = { path = "../aura-analysis" }
+# the winner_fingerprint lockstep guard's RBootstrap literal (moved from
+# aura-engine, #291/#292) — test-only, same coupling-guard rationale as the
+# other three dev-deps above.
+aura-backtest = { path = "../aura-backtest" }

crates/aura-bench/src/surfaces/campaign.rs

@@ -369,7 +369,8 @@ mod tests {
     #[test]
     fn winner_fingerprint_stays_in_lockstep_with_the_registry_record_types() {
         use aura_analysis::{FamilySelection, SelectionMode};
-        use aura_engine::{MetricStats, RBootstrap};
+        use aura_backtest::RBootstrap;
+        use aura_engine::MetricStats;
         use aura_registry::{CampaignRunRecord, CellRealization, StageRealization, StageSelection};
 
         let stats = MetricStats { mean: 0.1, p5: 0.0, p25: 0.05, p50: 0.1, p75: 0.15, p95: 0.2 };

crates/aura-campaign/Cargo.toml

@@ -12,8 +12,14 @@ publish.workspace = true
 [dependencies]
 # the scalar vocabulary: params cross the MemberRunner seam as (name, Scalar) pairs.
 aura-core = { path = "../aura-core" }
-# RunReport (the member result type) + the sweep/walk_forward orchestration machinery.
+# the sweep/walk_forward orchestration machinery (metric-agnostic; RunReport
+# itself is generic here — the M = RunMetrics instantiation is aura-backtest's).
 aura-engine = { path = "../aura-engine" }
+# aura-backtest carries the trading instantiation (M = RunMetrics: RunReport/
+# SweepFamily/SweepPoint/WindowRun aliases, RMetrics, r_bootstrap) — production
+# code uses it directly now (#291/#292, C28 phase 2), promoted from
+# [dev-dependencies].
+aura-backtest = { path = "../aura-backtest" }
 # FamilySelection — the selection-provenance type stage selections carry.
 aura-analysis = { path = "../aura-analysis" }
 # family + campaign-run stores and the optimize/optimize_plateau/optimize_deflated selectors.
@@ -28,9 +34,3 @@ serde_json = { workspace = true }
 # work-stealing pool the engine's run_indexed already uses — parallelism across
 # sims, never within one. Direct versioned dep, mirroring aura-engine.
 rayon = "1"
-
-[dev-dependencies]
-# aura-backtest supplies RunMetrics/summarize_r for the metric-vocabulary drift
-# guard's direct-source import (tests/metric_vocabulary_e2e.rs). Production
-# code reaches these types via the aura-engine re-export instead.
-aura-backtest = { path = "../aura-backtest" }

crates/aura-campaign/src/exec.rs

@@ -12,11 +12,11 @@ use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::Mutex;
 
 use aura_analysis::{FamilySelection, SelectionMode};
+use aura_backtest::{r_bootstrap, RunMetrics, RunReport, SweepFamily, SweepPoint, WindowRun};
 use aura_core::{zip_params, Cell, ParamSpec, Scalar, Timestamp};
 use aura_engine::{
-    r_bootstrap, sweep, walk_forward, GridSpace, ListSpace, RollMode, RunManifest, RunMetrics,
+    sweep, walk_forward, GridSpace, ListSpace, RollMode, RunManifest,
-    RunReport, Space, SweepFamily, SweepPoint, WalkForwardError, WindowBounds, WindowRoller,
+    Space, WalkForwardError, WindowBounds, WindowRoller,
-    WindowRun,
 };
 use aura_registry::{
     derive_trace_name, generalization, optimize, optimize_deflated, optimize_plateau,
@@ -1172,7 +1172,7 @@ fn run_walk_forward_stage(
 
     // Keep only real OOS reports (drop faulted-fold placeholders by the broker
     // sentinel) before building the family.
-    let surviving: Vec<&aura_engine::WindowOutcome> = result
+    let surviving: Vec<&aura_engine::WindowOutcome<RunMetrics>> = result
         .windows
         .iter()
         .filter(|w| w.run.oos_report.manifest.broker != FAULTED_MEMBER_PLACEHOLDER_BROKER)
@@ -1211,6 +1211,6 @@ enum WfStageOutcome {
 /// The surviving (non-faulted) walk-forward windows' OOS reports, in roll
 /// order — the same mapping `walkforward_member_reports` does over a full
 /// result, applied to the filtered survivor slice.
-fn walkforward_member_reports_of(windows: &[&aura_engine::WindowOutcome]) -> Vec<RunReport> {
+fn walkforward_member_reports_of(windows: &[&aura_engine::WindowOutcome<RunMetrics>]) -> Vec<RunReport> {
     windows.iter().map(|w| w.run.oos_report.clone()).collect()
 }

crates/aura-campaign/src/lib.rs

@@ -20,7 +20,7 @@ use std::collections::BTreeMap;
 use std::num::NonZeroUsize;
 
 use aura_core::Scalar;
-use aura_engine::RunReport;
+use aura_backtest::RunReport;
 use aura_registry::{check_r_metric, RegistryError};
 use aura_research::{Axis, CampaignDoc, Cmp, ProcessDoc, RiskRegime, SelectRule, StageBlock};
 
@@ -388,7 +388,8 @@ pub fn preflight(process: &ProcessDoc, campaign: &CampaignDoc) -> Result<(), Exe
 #[cfg(test)]
 mod tests {
     use super::*;
-    use aura_engine::{RMetrics, RunManifest, RunMetrics, Timestamp};
+    use aura_backtest::{RMetrics, RunMetrics};
+    use aura_engine::{RunManifest, Timestamp};
     use aura_research::SweepSelection;
 
     /// A report with every per-member scalar planted to a distinct value
@@ -991,8 +992,9 @@ mod wf_tests {
     use std::sync::Mutex;
 
     use aura_analysis::SelectionMode;
+    use aura_backtest::{RunMetrics, RunReport};
     use aura_core::{Scalar, ScalarKind};
-    use aura_engine::{RollMode, RunManifest, RunMetrics, RunReport, Timestamp, WindowRoller};
+    use aura_engine::{RollMode, RunManifest, Timestamp, WindowRoller};
     use aura_registry::{FamilyKind, Registry};
     use aura_research::{
         Axis, CampaignDoc, Cmp, DataSection, DocKind, DocRef, Predicate, Presentation,

crates/aura-campaign/tests/execute.rs

@@ -7,8 +7,9 @@ use std::collections::BTreeMap;
 use aura_campaign::{
     execute, CellSpec, ExecFault, MemberFault, MemberRunner, DEFAULT_PARALLEL_INSTRUMENTS,
 };
+use aura_backtest::{r_bootstrap, RMetrics, RunMetrics, RunReport};
 use aura_core::{Scalar, ScalarKind, Timestamp};
-use aura_engine::{r_bootstrap, RMetrics, RunManifest, RunMetrics, RunReport, SelectionMode};
+use aura_engine::{RunManifest, SelectionMode};
 use aura_registry::{generalization, FamilyKind, Registry, RegistryError, StageBootstrap};
 use aura_research::{
     Axis, CampaignDoc, Cmp, DataSection, DocKind, DocRef, Predicate, Presentation, ProcessDoc,

crates/aura-campaign/tests/member_seam_e2e.rs

@@ -12,7 +12,8 @@
 use std::collections::BTreeMap;
 
 use aura_campaign::{member_metric, CellSpec, MemberFault, MemberRunner, PER_MEMBER_METRICS, RANKABLE_METRICS};
-use aura_engine::{summarize, RMetrics, RunManifest, RunMetrics, RunReport, Scalar, Timestamp};
+use aura_backtest::{summarize, RMetrics, RunMetrics, RunReport};
+use aura_engine::{RunManifest, Scalar, Timestamp};
 
 fn cell() -> CellSpec {
     CellSpec {
@@ -28,7 +29,7 @@ fn cell() -> CellSpec {
 }
 
 /// A minimal external `MemberRunner`: ignores its params and produces
-/// metrics via the real `aura_engine::summarize` reduction (not a hand-typed
+/// metrics via the real `aura_backtest::summarize` reduction (not a hand-typed
 /// `RunMetrics` literal), echoing the `window_ms` it was called with into
 /// the manifest — exactly what a real consumer's harness-binding code does.
 struct FixedRunner;
@@ -64,7 +65,7 @@ impl MemberRunner for FixedRunner {
 /// `&dyn MemberRunner` (the shape a heterogeneous, per-consumer roster of
 /// runners requires); the exact sub-window given at the call site — not
 /// `cell.window_ms`, a walk-forward stage passes a narrower sub-range —
-/// reaches the runner unmodified; and the real `aura_engine::summarize()`
+/// reaches the runner unmodified; and the real `aura_backtest::summarize()`
 /// output the runner produced is what `member_metric` reads back through the
 /// crate's public per-name resolver.
 #[test]

crates/aura-cli/src/campaign_run.rs

@@ -19,11 +19,12 @@ use std::num::NonZeroUsize;
 use std::path::{Path, PathBuf};
 use std::sync::{mpsc, Arc};
 
+use aura_backtest::{summarize, summarize_r, RunReport};
 use aura_campaign::{CampaignOutcome, CellOutcome, CellSpec, ExecFault, MemberFault, MemberRunner};
 use aura_core::{Cell, ParamSpec, PrimitiveBuilder, Scalar, ScalarKind, Timestamp};
 use aura_engine::{
-    blueprint_from_json, f64_field, summarize, summarize_r, ColumnarTrace, Composite,
+    blueprint_from_json, f64_field, ColumnarTrace, Composite,
-    FamilySelection, RunReport,
+    FamilySelection,
 };
 use aura_ingest::{instrument_geometry, open_columns_window, unix_ms_to_epoch_ns};
 use aura_registry::{CampaignRunRecord, WriteKind};

crates/aura-cli/src/main.rs

@@ -16,13 +16,13 @@ use render::{ChartData, ChartMeta, ChartMode, ReduceKind, Series};
 use aura_core::{zip_params, Cell, Firing, ParamSpec, PrimitiveBuilder, Scalar, ScalarKind, Timestamp};
 use aura_composites::{cost_graph, risk_executor, StopRule};
 use aura_engine::{
-    blueprint_from_json, blueprint_to_json, f64_field, join_on_ts, monte_carlo, param_stability,
+    blueprint_from_json, blueprint_to_json, f64_field, join_on_ts, param_stability,
-    r_metrics_from_rs, summarize, summarize_r, walk_forward, window_of, BindError, BlueprintNode,
+    walk_forward, window_of, BindError, BlueprintNode,
-    Composite, FamilySelection, GraphBuilder, Harness, JoinedRow, McAggregate,
+    Composite, FamilySelection, GraphBuilder, Harness, JoinedRow,
-    McFamily, MeasurementReport, RBootstrap, RollMode, RunManifest, RunMetrics, RunReport,
+    MeasurementReport, RollMode, RunManifest,
     SelectionMode,
-    SweepFamily, SweepPoint, SyntheticSpec, VecSource, WalkForwardResult,
+    SyntheticSpec, VecSource,
-    WindowBounds, WindowRoller, WindowRun,
+    WindowBounds, WindowRoller,
 };
 use aura_registry::{
     check_r_metric, group_families, mc_member_reports, optimize_deflated,
@@ -30,7 +30,11 @@ use aura_registry::{
     FamilyMember, Generalization, NameKind, PlateauMode, Registry, RunTraces,
     DEFLATION_BLOCK_LEN, DEFLATION_N_RESAMPLES,
 };
-use aura_backtest::{SimBroker, PM_FIELD_NAMES, PM_RECORD_KINDS};
+use aura_backtest::{
+    monte_carlo, r_metrics_from_rs, summarize, summarize_r, McAggregate, McFamily, RBootstrap,
+    RunMetrics, RunReport, SimBroker, SweepFamily, SweepPoint, WalkForwardResult, WindowRun,
+    PM_FIELD_NAMES, PM_RECORD_KINDS,
+};
 use aura_std::{
     GatedRecorder, LinComb, Recorder, RollingMax, RollingMin, SeriesReducer, Sub,
 };
@@ -6347,7 +6351,7 @@ mod tests {
         // breaks a test instead of silently shipping. Pins the user-visible wire
         // shape of the mc R-bootstrap render (the parser + engine primitive are
         // covered elsewhere; this is the output-shape layer).
-        let boot = aura_engine::r_bootstrap(&[1.0, -0.5, 2.0, -1.0], 64, 2, 7);
+        let boot = aura_backtest::r_bootstrap(&[1.0, -0.5, 2.0, -1.0], 64, 2, 7);
         let line = mc_r_bootstrap_json(&boot);
         let v: serde_json::Value = serde_json::from_str(&line).expect("canonical json line");
         let obj = &v["mc_r_bootstrap"];

crates/aura-cli/src/verb_sugar.rs

@@ -363,7 +363,7 @@ pub(crate) fn run_sweep_sugar(
 /// itself enforces.
 fn cross_instrument_members(
     outcome: &aura_campaign::CampaignOutcome,
-) -> Vec<aura_engine::RunReport> {
+) -> Vec<aura_backtest::RunReport> {
     outcome
         .cells
         .iter()
@@ -1180,7 +1180,7 @@ mod tests {
     /// A minimal `RunReport` stamped with `instrument`, distinguishable in a
     /// members list by nothing but that stamp — the field
     /// `cross_instrument_members` preserves verbatim from a nominating cell.
-    fn fake_report(instrument: &str) -> aura_engine::RunReport {
+    fn fake_report(instrument: &str) -> aura_backtest::RunReport {
         aura_engine::RunReport {
             manifest: aura_engine::RunManifest {
                 commit: String::new(),
@@ -1194,7 +1194,7 @@ mod tests {
                 topology_hash: None,
                 project: None,
             },
-            metrics: aura_engine::RunMetrics {
+            metrics: aura_backtest::RunMetrics {
                 total_pips: 0.0,
                 max_drawdown: 0.0,
                 bias_sign_flips: 0,

crates/aura-composites/tests/risk_executor.rs

@@ -9,8 +9,8 @@ use aura_core::{
     ScalarKind, Timestamp,
 };
 use aura_composites::{risk_executor, StopRule};
-use aura_engine::{summarize_r, GraphBuilder, RunMetrics, VecSource};
+use aura_engine::{GraphBuilder, VecSource};
-use aura_backtest::{PM_FIELD_NAMES, PM_RECORD_KINDS};
+use aura_backtest::{summarize_r, RunMetrics, PM_FIELD_NAMES, PM_RECORD_KINDS};
 use aura_std::Recorder;
 use std::sync::mpsc::channel;
 
@@ -225,7 +225,7 @@ fn risk_executor_vol_tf_stop_arm_bootstraps_and_folds() {
 /// folding the dense R-record. `exec` is the variant under test (bound or open); `params`
 /// is the positional point the open variant needs (empty for the bound one). Shared by the
 /// open-vs-bound equivalence test below so the two arms run byte-identical inputs.
-fn fold_executor(exec: aura_engine::Composite, params: Vec<Scalar>) -> aura_engine::RMetrics {
+fn fold_executor(exec: aura_engine::Composite, params: Vec<Scalar>) -> aura_backtest::RMetrics {
     let (tx, rx) = channel();
     let mut g = GraphBuilder::new("vol_open_harness");
     let price = g.source_role("price", ScalarKind::F64);

crates/aura-engine/Cargo.toml

@@ -12,11 +12,6 @@ aura-core = { path = "../aura-core" }
 # (#136, reduced to the domain-free half by #291). Re-exported via `report::`
 # so callers resolve the types through aura-engine unchanged (C18).
 aura-analysis = { path = "../aura-analysis" }
-# aura-backtest holds the backtest metric reductions re-exported through
-# `report::` (#291). TRANSIENT widening of the known C28 layer violation: the
-# phase-2 edge cut (#147) removes this production edge again on the same
-# branch.
-aura-backtest = { path = "../aura-backtest" }
 # serde is admitted under the amended C16 per-case dependency policy (INDEX.md):
 # it gives the run-report types a typed (de)serialization path for the run
 # registry (cycle 0029). serde's output is deterministic (C1-safe).
@@ -34,6 +29,12 @@ serde_json = { workspace = true }
 rayon = "1"
 
 [dev-dependencies]
+# aura-backtest is TEST-ONLY (back to its pre-#291 home, C28 phase 2 D3): the
+# engine's own tests instantiate `RunReport<M>` with `M = RunMetrics`
+# (SimBroker fixtures, summarize/RunMetrics literals) via aura-backtest's
+# concrete aliases. The library code stays metric-agnostic — `RunReport<M>`
+# names no concrete metric type — so aura-backtest is out of [dependencies].
+aura-backtest = { path = "../aura-backtest" }
 # aura-std is a TEST-ONLY dependency: the engine's own integration tests
 # (r_sma_e2e, random_sweep_e2e, ger40_breakout) drive real standard nodes through
 # a bootstrapped Harness. The engine's library code never names a concrete node — it

crates/aura-engine/src/blueprint.rs

@@ -560,9 +560,10 @@ impl SweepBinder {
     /// Resolve the named axes against `param_space()` into a positional grid and
     /// run the disjoint sweep. `sweep` is [`SweepBinder::sweep_with_lattice`] with
     /// the lattice dropped, so every existing caller is byte-unchanged.
-    pub fn sweep<F>(self, run_one: F) -> Result<SweepFamily, BindError>
+    pub fn sweep<M, F>(self, run_one: F) -> Result<SweepFamily<M>, BindError>
     where
-        F: Fn(&[Cell]) -> RunReport + Sync,
+        M: Send,
+        F: Fn(&[Cell]) -> RunReport<M> + Sync,
     {
         self.sweep_with_lattice(run_one).map(|(family, _lattice)| family)
     }
@@ -571,9 +572,10 @@ impl SweepBinder {
     /// in `param_space()` / odometer order). The lattice is what a plateau
     /// neighbourhood walk needs (cycle 0077); only the engine's post-`resolve_axes`
     /// grid holds it in the correct order.
-    pub fn sweep_with_lattice<F>(self, run_one: F) -> Result<(SweepFamily, Vec<usize>), BindError>
+    pub fn sweep_with_lattice<M, F>(self, run_one: F) -> Result<(SweepFamily<M>, Vec<usize>), BindError>
     where
-        F: Fn(&[Cell]) -> RunReport + Sync,
+        M: Send,
+        F: Fn(&[Cell]) -> RunReport<M> + Sync,
     {
         let space = self.bp.param_space();
         check_param_namespace_injective(&space).map_err(BindError::Compile)?;
@@ -604,9 +606,10 @@ impl RandomBinder {
     /// Resolve the named ranges against `param_space()` into a positional
     /// `RandomSpace` and run the disjoint sweep. `count`/`seed` are `RandomSpace`'s
     /// extra inputs — the only signature difference from [`SweepBinder::sweep`].
-    pub fn sweep<F>(self, count: usize, seed: u64, run_one: F) -> Result<SweepFamily, BindError>
+    pub fn sweep<M, F>(self, count: usize, seed: u64, run_one: F) -> Result<SweepFamily<M>, BindError>
     where
-        F: Fn(&[Cell]) -> RunReport + Sync,
+        M: Send,
+        F: Fn(&[Cell]) -> RunReport<M> + Sync,
     {
         let space = self.bp.param_space();
         check_param_namespace_injective(&space).map_err(BindError::Compile)?;
@@ -1382,7 +1385,8 @@ fn slot_kind(t: Target, flat_signatures: &[NodeSchema]) -> Result<ScalarKind, Co
 mod tests {
     use super::*;
     use crate::test_fixtures::{composite_sma_cross_harness, synthetic_prices};
-    use crate::{f64_field, summarize, ParamRange, RandomSpace, RunManifest, VecSource};
+    use crate::{f64_field, ParamRange, RandomSpace, RunManifest, VecSource};
+    use aura_backtest::{summarize, RunMetrics};
     use aura_core::{Cell, Ctx, FieldSpec, Firing, NodeSchema, Timestamp};
     use aura_backtest::SimBroker;
     use aura_std::{Ema, LinComb, Recorder, Sma, Sub};
@@ -1858,7 +1862,7 @@ mod tests {
     /// `fn` (Copy + Sync) so it serves both as the `sweep`/binder closure and as a
     /// direct reference. Deterministic, so the same point reproduces its report
     /// exactly — that is what makes two families comparable for equality.
-    fn run_point(point: &[Cell]) -> RunReport {
+    fn run_point(point: &[Cell]) -> RunReport<RunMetrics> {
         let (bp, rx_eq, rx_ex) = composite_sma_cross_harness();
         let mut h = bp
             .bootstrap_with_cells(point)
@@ -2038,7 +2042,7 @@ mod tests {
             .axis("sma_cross.fast.length", [2.0, 3.0]) // F64 values for the I64 slot
             .axis("sma_cross.slow.length", [4])
             .axis("bias.scale", [0.5])
-            .sweep(|_: &[Cell]| -> RunReport { panic!("axis pre-validation must reject before running") });
+            .sweep(|_: &[Cell]| -> RunReport<RunMetrics> { panic!("axis pre-validation must reject before running") });
         assert_eq!(
             result,
             Err(BindError::KindMismatch {

crates/aura-engine/src/harness.rs

@@ -21,6 +21,7 @@
 //! sources are four cycles, so RustAst's `cycle_id`-equality barrier could never
 //! fire across sources — the timestamp generalizes it faithfully.
 
+pub use aura_analysis::SplitMix64;
 use aura_core::{AnyColumn, Cell, Ctx, Firing, Node, NodeSchema, Scalar, ScalarKind, Timestamp};
 
 /// Forwards one field (`from_field`) of a producer's output record into a
@@ -220,31 +221,6 @@ pub fn window_of(sources: &[Box<dyn Source>]) -> Option<(Timestamp, Timestamp)>
     acc
 }
 
-/// A tiny, fully-deterministic, dependency-free PRNG (SplitMix64). The seed
-/// completely determines the sequence; no external entropy, no global state.
-/// Bit-stable across toolchains and crate versions — the property C1 needs for
-/// seed-as-input reproducibility (C12).
-pub struct SplitMix64 {
-    state: u64,
-}
-
-impl SplitMix64 {
-    pub fn new(seed: u64) -> Self {
-        Self { state: seed }
-    }
-    pub fn next_u64(&mut self) -> u64 {
-        self.state = self.state.wrapping_add(0x9E37_79B9_7F4A_7C15);
-        let mut z = self.state;
-        z = (z ^ (z >> 30)).wrapping_mul(0xBF58_476D_1CE4_E5B9);
-        z = (z ^ (z >> 27)).wrapping_mul(0x94D0_49BB_1331_11EB);
-        z ^ (z >> 31)
-    }
-    /// A uniform `f64` in `[0, 1)` from the top 53 bits.
-    pub(crate) fn next_f64(&mut self) -> f64 {
-        (self.next_u64() >> 11) as f64 / ((1u64 << 53) as f64)
-    }
-}
-
 /// Config for a seeded synthetic price walk — everything *except* the seed.
 /// Partially applying it (`|seed| spec.source(seed)`) yields the
 /// `Fn(u64) -> impl Source` contract the Monte-Carlo family (#68) consumes

crates/aura-engine/src/lib.rs

@@ -13,28 +13,27 @@
 //! - [`BootstrapError`] — wiring faults caught once, at bootstrap (kind
 //!   mismatch, bad index, directed cycle).
 //!
-//! Delivered in cycle 0009 — the run report surface:
+//! Delivered in cycle 0009, generic since #292 — the run-record surface:
 //!
-//! - [`RunMetrics`] / [`RunManifest`] / [`RunReport`] — the `(manifest, metrics)`
+//! - [`RunReport`] / [`RunManifest`] — the `(manifest, metrics)` pair C18
-//!   pair C18 mandates per run, with [`RunReport::to_json`] for the structured
+//!   mandates per run. [`RunReport`] is generic over its metric payload `M`:
-//!   C14 face;
+//!   the engine names no concrete metric type; the backtest layer
-//! - [`summarize`] — the post-run pure reduction over a run's recorded
+//!   (`aura-backtest`) instantiates it and hosts the pip/R reductions. C14 face
-//!   pip-equity + exposure streams into [`RunMetrics`]; [`f64_field`] bridges a
+//!   via [`RunReport::to_json`]; [`f64_field`] bridges a recording sink's
-//!   recording sink's `Vec<Scalar>` rows to it.
+//!   `Vec<Scalar>` rows into the `(ts, f64)` streams a metric reduction folds.
 //!
 //! Orchestration families of the atomic sim unit
-//! (`(topology + params + data-window + seed)`) ship along three of C12's axes: the
+//! (`(topology + params + data-window + seed)`) ship along C12's axes: the
-//! **grid** axis via [`sweep`] over a [`GridSpace`] into a [`SweepFamily`], the
+//! **grid** axis via [`sweep`] over a [`GridSpace`] into a [`SweepFamily`], and
-//! **window** axis via [`walk_forward`] over a [`WindowRoller`] into a
+//! the **window** axis via [`walk_forward`] over a [`WindowRoller`] into a
-//! [`WalkForwardResult`] (C12 axis 3: rolling in-sample/out-of-sample splits, the
+//! [`WalkForwardResult`] (C12 axis 3: rolling in-sample/out-of-sample splits,
-//! in-sample optimize closure-supplied), and
+//! the in-sample optimize closure-supplied). The **seed** axis (C12 axis 4:
-//! the **seed** axis via [`monte_carlo`] over a seed set into an [`McFamily`]
+//! Monte-Carlo *is* a sweep over seeds) is assembled one rung out in
-//! (C12 axis 4: Monte-Carlo *is* a sweep over seeds; both drive one shared
+//! `aura-backtest`, over the same shared disjoint-parallel executor this crate
-//! disjoint-parallel executor). The **seed** input itself ships too:
+//! provides. The **seed** input itself ships here: [`SyntheticSpec`] is a
-//! [`SyntheticSpec`] is a seeded `Source` producer (`Fn(u64) -> impl Source`,
+//! seeded `Source` producer (`Fn(u64) -> impl Source`, C12 seed-as-input) whose
-//! C12 seed-as-input) whose stream is fully seed-determined, making
+//! stream is fully seed-determined, making [`RunManifest`]'s seed a live
-//! [`RunManifest`]'s seed a live captured input; the atomic unit's `-> metrics`
+//! captured input.
-//! reduction ships via [`summarize`].
 //!
 //! Still to come (subsequent cycles): the broker-independent position-event
 //! output and downstream broker nodes (C10), the random param-sweep
@@ -48,7 +47,6 @@ mod builder;
 mod construction;
 mod graph_model;
 mod harness;
-mod mc;
 mod report;
 mod sweep;
 mod walkforward;
@@ -69,19 +67,19 @@ pub use harness::{
     SourceSpec, SplitMix64, SyntheticSpec, TapBindError, Target, VecSource,
 };
 pub use report::{
-    derive_position_events, expected_max_of_normals, f64_field, inv_norm_cdf, join_on_ts,
+    expected_max_of_normals, f64_field, inv_norm_cdf, join_on_ts,
-    r_metrics_from_rs, summarize, summarize_r, ColumnarTrace, FamilySelection, JoinedRow,
+    ColumnarTrace, FamilySelection, JoinedRow,
-    MeasurementReport, PositionAction, PositionEvent, ProjectProvenance, RMetrics, RunManifest,
+    MeasurementReport, ProjectProvenance, RunManifest,
-    RunMetrics, RunReport, SelectionMode,
+    RunReport, SelectionMode,
 };
 pub use sweep::{
-    sweep, GridSpace, ListSpace, ParamRange, RandomSpace, Space, SweepError, SweepFamily,
+    run_indexed, sweep, GridSpace, ListSpace, ParamRange, RandomSpace, Space, SweepError,
-    SweepPoint,
+    SweepFamily, SweepPoint,
-};
-pub use mc::{
-    monte_carlo, r_bootstrap, resample_block, McAggregate, McDraw, McFamily, MetricStats,
-    RBootstrap,
 };
+// `resample_block`/`MetricStats` are foundation-grade (aura-analysis, #291);
+// re-exported here (unchanged exported name set) so existing
+// `aura_engine::{resample_block, MetricStats}` consumers stay source-compatible.
+pub use aura_analysis::{resample_block, MetricStats};
 pub use walkforward::{
     param_stability, walk_forward, RollMode, WalkForwardError, WalkForwardResult,
     WindowBounds, WindowOutcome, WindowRoller, WindowRun,

crates/aura-engine/src/report.rs

@@ -8,21 +8,17 @@
 //! cycle 0029) — the same encoder the run registry uses, so a record's stdout
 //! and on-disk shapes coincide.
 
-use aura_core::{Scalar, ScalarKind, SeriesFold, Timestamp};
+use aura_core::{Scalar, ScalarKind, Timestamp};
 use std::collections::HashMap;
 
-// The backtest reductions (R-metrics, the position-event table) live in
+// The backtest reductions (R-metrics, the position-event table, `summarize`)
-// `aura-backtest::metrics`, the domain-free hurdle math + selection provenance
+// moved to `aura-backtest::metrics` (C28 phase 2, #291/#292 — the engine's
-// in `aura-analysis` (issues #136, #291); both are re-imported here so
+// production edge to aura-backtest is cut, D4); the domain-free hurdle math +
+// selection provenance stays re-imported here from `aura-analysis` so
 // `report::`'s namespace — and therefore `aura-engine`'s `lib.rs` re-export and
-// every `crate::report::X` reference — resolves unchanged (C18 byte-identity).
+// every `crate::report::X` reference — resolves unchanged for those names
-// `summarize` (below) bridges trace columns into these types at the engine
+// (C18 byte-identity).
-// boundary, so it stays here.
 pub use aura_analysis::{expected_max_of_normals, inv_norm_cdf, FamilySelection, SelectionMode};
-pub use aura_backtest::{
-    derive_position_events, r_metrics_from_rs, summarize_r, PositionAction,
-    PositionEvent, RMetrics, RunMetrics,
-};
 
 /// The reproducible run descriptor (C18). **Caller-supplied**: the engine
 /// cannot introspect a git commit, an RNG seed, or a broker label — the World
@@ -92,16 +88,18 @@ pub struct ProjectProvenance {
     pub commit: Option<String>,
 }
 
-/// A run's full structured result: the descriptor plus the metrics it
+/// A run's full structured result: the descriptor plus the metric payload `M`
-/// reproduces. The durable run record of C18 ("stores manifests + metrics,
+/// it reproduces. Metric-agnostic (C28): the engine never names a concrete
-/// re-derives full results on demand").
+/// metric type; the backtest layer instantiates `M` (see `aura-backtest`). The
+/// durable run record of C18 ("stores manifests + metrics, re-derives full
+/// results on demand").
 #[derive(Clone, Debug, PartialEq, serde::Serialize, serde::Deserialize)]
-pub struct RunReport {
+pub struct RunReport<M> {
     pub manifest: RunManifest,
-    pub metrics: RunMetrics,
+    pub metrics: M,
 }
 
-impl RunReport {
+impl<M: serde::Serialize> RunReport<M> {
     /// Render the canonical, machine-readable JSON (C14) via serde — the same
     /// encoder the run registry uses on disk, so a record's stdout shape and its
     /// `runs.jsonl` shape are byte-identical. `params` is an array of
@@ -130,35 +128,9 @@ impl MeasurementReport {
     }
 }
 
-/// Reduce a run's recorded pip-equity + exposure streams into summary metrics.
+/// Bridge a recording sink's recorded `(ts, row)` stream to a metric reduction
-/// Pure — identical inputs yield identical metrics (C1/C12). Timestamps are
+/// (e.g. `aura-backtest`'s `summarize`): extract one `f64` field of each row
-/// carried in the input to match exactly what a sink records; the reduction
+/// into `(ts, f64)` samples. Panics if a
-/// itself is value-only (it does not read the timestamps).
-pub fn summarize(
-    equity: &[(Timestamp, f64)],
-    exposure: &[(Timestamp, f64)],
-) -> RunMetrics {
-    // total_pips + max_drawdown fold over equity; sign_flips folds over exposure.
-    // SeriesFold is the single arithmetic source of truth (shared with the
-    // SeriesReducer sink), so this is byte-identical to the prior inline loops.
-    let mut eqf = SeriesFold::new();
-    for &(_, v) in equity {
-        eqf.fold(v);
-    }
-    let mut exf = SeriesFold::new();
-    for &(_, v) in exposure {
-        exf.fold(v);
-    }
-    RunMetrics {
-        total_pips: eqf.last,
-        max_drawdown: eqf.max_drawdown,
-        bias_sign_flips: exf.sign_flips,
-        r: None,
-    }
-}
-
-/// Bridge a recording sink's recorded `(ts, row)` stream to [`summarize`]:
-/// extract one `f64` field of each row into `(ts, f64)` samples. Panics if a
 /// row has no such field or the field is not an `f64` scalar — a wiring bug (a
 /// sink's declared kinds are fixed at bootstrap, so a correctly-wired
 /// equity/exposure sink always yields `f64` at field 0), surfaced like the
@@ -312,12 +284,7 @@ fn scalar_to_f64(s: Scalar) -> f64 {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::{Edge, FlatGraph, Harness, SourceSpec, Target, VecSource};
+    use aura_backtest::RunMetrics;
-    use aura_core::{Firing, NodeSchema, PortSpec, ScalarKind};
-    use aura_backtest::SimBroker;
-    use aura_std::{Recorder, Sma, Sub};
-    use aura_strategy::Bias;
-    use std::sync::mpsc;
 
     #[test]
     fn runmanifest_without_selection_field_deserialises_to_none() {
@@ -472,172 +439,6 @@ mod tests {
         assert_eq!(sel.n_neighbours, None);
     }
 
-    /// The declared signature of a `Recorder` over one f64 column (the sink shape
-    /// the two-sink harness uses).
-    fn f64_recorder_sig() -> NodeSchema {
-        NodeSchema {
-            inputs: vec![PortSpec { kind: ScalarKind::F64, firing: Firing::Any, name: "in".into() }],
-            output: vec![],
-            params: vec![],
-        }
-    }
-
-    /// Build an f64 source stream from (timestamp, value) points (mirrors the
-    /// harness.rs test helper; the e2e test needs its own copy — the harness
-    /// test module's is private to that module).
-    fn f64_stream(points: &[(i64, f64)]) -> Vec<(Timestamp, Scalar)> {
-        points.iter().map(|&(t, v)| (Timestamp(t), Scalar::f64(v))).collect()
-    }
-
-    /// Bootstrap the cycle-0007 signal-quality harness with TWO sinks: one on
-    /// the SimBroker equity output (node 4 -> node 5) and one on the Bias
-    /// output (node 3 -> node 6). Returns the harness plus the two receivers.
-    #[allow(clippy::type_complexity)]
-    fn build_two_sink_harness() -> (
-        Harness,
-        mpsc::Receiver<(Timestamp, Vec<Scalar>)>,
-        mpsc::Receiver<(Timestamp, Vec<Scalar>)>,
-    ) {
-        let (tx_eq, rx_eq) = mpsc::channel();
-        let (tx_ex, rx_ex) = mpsc::channel();
-        let h = Harness::bootstrap(FlatGraph {
-            nodes: vec![
-                Box::new(Sma::new(2)),                                       // 0
-                Box::new(Sma::new(4)),                                       // 1
-                Box::new(Sub::new()),                                        // 2
-                Box::new(Bias::new(0.5)),                                    // 3
-                Box::new(SimBroker::new(0.0001)),                            // 4
-                Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx_eq)), // 5 equity sink
-                Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx_ex)), // 6 exposure sink
-            ],
-            signatures: vec![
-                Sma::builder().schema().clone(),
-                Sma::builder().schema().clone(),
-                Sub::builder().schema().clone(),
-                Bias::builder().schema().clone(),
-                SimBroker::builder(0.0001).schema().clone(),
-                f64_recorder_sig(),
-                f64_recorder_sig(),
-            ],
-            sources: vec![SourceSpec::raw(ScalarKind::F64, vec![
-                    Target { node: 0, slot: 0 },
-                    Target { node: 1, slot: 0 },
-                    Target { node: 4, slot: 1 }, // price into the broker
-                ])],
-            edges: vec![
-                Edge { from: 0, to: 2, slot: 0, from_field: 0 },
-                Edge { from: 1, to: 2, slot: 1, from_field: 0 },
-                Edge { from: 2, to: 3, slot: 0, from_field: 0 },
-                Edge { from: 3, to: 4, slot: 0, from_field: 0 },
-                Edge { from: 4, to: 5, slot: 0, from_field: 0 }, // equity -> sink 5
-                Edge { from: 3, to: 6, slot: 0, from_field: 0 }, // exposure -> sink 6
-            ],
-            taps: Vec::new(),
-        })
-        .expect("valid signal-quality DAG");
-        (h, rx_eq, rx_ex)
-    }
-
-    fn run_once() -> RunReport {
-        let (mut h, rx_eq, rx_ex) = build_two_sink_harness();
-        h.run(vec![Box::new(VecSource::new(f64_stream(&[
-            (1, 1.0000),
-            (2, 1.0010),
-            (3, 1.0025),
-            (4, 1.0020),
-            (5, 1.0040),
-        ])))]);
-        let eq_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_eq.try_iter().collect();
-        let ex_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_ex.try_iter().collect();
-        let equity = f64_field(&eq_rows, 0);
-        let exposure = f64_field(&ex_rows, 0);
-        let metrics = summarize(&equity, &exposure);
-        RunReport {
-            manifest: RunManifest {
-                commit: "test-commit".to_string(),
-                params: vec![
-                    ("sma_fast".to_string(), Scalar::i64(2)),
-                    ("sma_slow".to_string(), Scalar::i64(4)),
-                    ("bias_scale".to_string(), Scalar::f64(0.5)),
-                ],
-                defaults: vec![],
-                window: (Timestamp(1), Timestamp(5)),
-                seed: 0,
-                broker: "sim-optimal(pip_size=0.0001)".to_string(),
-                selection: None,
-                instrument: None,
-                topology_hash: None,
-                project: None,
-            },
-            metrics,
-        }
-    }
-
-    #[test]
-    fn report_is_deterministic_end_to_end() {
-        let r1 = run_once();
-        let r2 = run_once();
-        // a run actually emitted metrics over a non-empty pip curve
-        assert!(r1.metrics.total_pips.is_finite());
-        // same manifest -> same metrics (C1/C12): two runs are bit-identical
-        assert_eq!(r1.metrics, r2.metrics);
-        assert_eq!(r1.to_json(), r2.to_json());
-    }
-
-    fn samples(values: &[f64]) -> Vec<(Timestamp, f64)> {
-        values
-            .iter()
-            .enumerate()
-            .map(|(i, &v)| (Timestamp(i as i64 + 1), v))
-            .collect()
-    }
-
-    #[test]
-    fn summarize_total_pips_is_last_cumulative_value() {
-        let equity = samples(&[0.0, 5.0, 4.0, 12.0]);
-        let m = summarize(&equity, &[]);
-        assert_eq!(m.total_pips, 12.0);
-    }
-
-    #[test]
-    fn summarize_is_zero_on_empty_streams() {
-        let m = summarize(&[], &[]);
-        assert_eq!(m.total_pips, 0.0);
-        assert_eq!(m.max_drawdown, 0.0);
-        assert_eq!(m.bias_sign_flips, 0);
-    }
-
-    #[test]
-    fn summarize_max_drawdown_is_worst_peak_to_trough() {
-        // peak 10 then trough 5 (drop 5), recovers to 8; worst drop is 5,
-        // not the final drop (10 -> 8 = 2).
-        let equity = samples(&[0.0, 10.0, 5.0, 8.0]);
-        let m = summarize(&equity, &[]);
-        assert_eq!(m.max_drawdown, 5.0);
-    }
-
-    #[test]
-    fn summarize_max_drawdown_zero_on_monotonic_curve() {
-        let equity = samples(&[0.0, 1.0, 2.0, 3.0]);
-        let m = summarize(&equity, &[]);
-        assert_eq!(m.max_drawdown, 0.0);
-    }
-
-    #[test]
-    fn summarize_sign_flips_counts_signum_changes() {
-        // signum series: + + - 0 -  ->  flips at +->-, -->0, 0->- = 3.
-        let exposure = samples(&[0.5, 0.5, -0.5, 0.0, -0.5]);
-        let m = summarize(&[], &exposure);
-        assert_eq!(m.bias_sign_flips, 3);
-    }
-
-    #[test]
-    fn summarize_sign_flips_zero_on_constant_sign() {
-        let exposure = samples(&[0.2, 0.5, 1.0, 0.7]);
-        let m = summarize(&[], &exposure);
-        assert_eq!(m.bias_sign_flips, 0);
-    }
-
     #[test]
     fn f64_field_projects_the_named_field() {
         let rows = vec![
@@ -739,7 +540,7 @@ mod tests {
         let json = serde_json::to_string(&report).expect("serialize RunReport");
         // window is a 2-element [from, to] array (Timestamp newtype is transparent)
         assert!(json.contains("\"window\":[1,6]"), "window shape: {json}");
-        let back: RunReport = serde_json::from_str(&json).expect("deserialize RunReport");
+        let back: RunReport<RunMetrics> = serde_json::from_str(&json).expect("deserialize RunReport");
         assert_eq!(back, report);
     }
 

crates/aura-engine/src/sweep.rs

@@ -344,20 +344,20 @@ impl Space for ListSpace {
 /// and `report` carries the run's `(manifest, metrics)` — the unit the run
 /// registry indexes (C18).
 #[derive(Clone, Debug, PartialEq)]
-pub struct SweepPoint {
+pub struct SweepPoint<M> {
     pub params: Vec<Cell>,
-    pub report: RunReport,
+    pub report: RunReport<M>,
 }
 
 /// The ordered result family of a sweep — one `SweepPoint` per grid point, in
 /// enumeration (odometer) order, independent of thread completion order.
 #[derive(Clone, Debug, PartialEq)]
-pub struct SweepFamily {
+pub struct SweepFamily<M> {
     pub space: Vec<ParamSpec>,
-    pub points: Vec<SweepPoint>,
+    pub points: Vec<SweepPoint<M>>,
 }
 
-impl SweepFamily {
+impl<M> SweepFamily<M> {
     /// The i-th point's params paired with their names — a derived view over the
     /// carried param-space (reuses [`zip_params`]); no new per-point state.
     pub fn named_params(&self, i: usize) -> Vec<(String, Scalar)> {
@@ -372,9 +372,10 @@ impl SweepFamily {
 /// enumeration order. Kept as one source of truth so the two callers can never
 /// drift on assembly logic (grid points, `RandomSpace` draws) while their pool
 /// selection differs.
-fn assemble_sweep<F>(space: Vec<ParamSpec>, points: Vec<Vec<Cell>>, run_one: &F) -> SweepFamily
+fn assemble_sweep<M, F>(space: Vec<ParamSpec>, points: Vec<Vec<Cell>>, run_one: &F) -> SweepFamily<M>
 where
-    F: Fn(&[Cell]) -> RunReport + Sync,
+    M: Send,
+    F: Fn(&[Cell]) -> RunReport<M> + Sync,
 {
     let reports = run_indexed(points.len(), |i| run_one(&points[i]));
     SweepFamily {
@@ -395,10 +396,11 @@ where
 /// shared rayon pool (`run_indexed`): a nested sweep / walk-forward shares its
 /// workers rather than each spawning its own `available_parallelism()`
 /// OS-thread batch.
-pub fn sweep<S, F>(space: &S, run_one: F) -> SweepFamily
+pub fn sweep<S, M, F>(space: &S, run_one: F) -> SweepFamily<M>
 where
     S: Space,
-    F: Fn(&[Cell]) -> RunReport + Sync,
+    M: Send,
+    F: Fn(&[Cell]) -> RunReport<M> + Sync,
 {
     let points = space.points();
     assemble_sweep(space.param_specs().to_vec(), points, &run_one)
@@ -421,7 +423,7 @@ where
 // passing `run_one` directly would move it into `Map` and require `F: Send`, a
 // bound the `_with_threads` wrappers' `install`-borrowed callers don't carry.
 #[allow(clippy::redundant_closure)]
-pub(crate) fn run_indexed<T, F>(n: usize, run_one: F) -> Vec<T>
+pub fn run_indexed<T, F>(n: usize, run_one: F) -> Vec<T>
 where
     T: Send,
     F: Fn(usize) -> T + Sync,
@@ -441,10 +443,11 @@ where
 /// *borrows* `run_one` and moves the already-owned `space`/`points`, so no
 /// `Send` bound on `F` is needed.
 #[cfg(test)]
-fn sweep_with_threads<S, F>(space: &S, nthreads: usize, run_one: F) -> SweepFamily
+fn sweep_with_threads<S, M, F>(space: &S, nthreads: usize, run_one: F) -> SweepFamily<M>
 where
     S: Space,
-    F: Fn(&[Cell]) -> RunReport + Sync,
+    M: Send,
+    F: Fn(&[Cell]) -> RunReport<M> + Sync,
 {
     let points = space.points();
     let param_specs = space.param_specs().to_vec();
@@ -459,7 +462,8 @@ where
 mod tests {
     use super::*;
     use crate::test_fixtures::{composite_sma_cross_harness, synthetic_prices};
-    use crate::{f64_field, summarize, RunManifest, VecSource};
+    use crate::{f64_field, RunManifest, VecSource};
+    use aura_backtest::{summarize, RunMetrics};
     use aura_core::{Cell, ParamSpec, Scalar, ScalarKind, Timestamp};
 
     fn i64_space(n: usize) -> Vec<ParamSpec> {
@@ -720,7 +724,7 @@ mod tests {
     /// AND a direct reference for the "sweep == N independent runs" comparison.
     /// The manifest is a minimal fixed fixture — the metrics are the run's, and
     /// determinism makes `run_point` reproduce a point's report exactly.
-    fn run_point(point: &[Cell]) -> RunReport {
+    fn run_point(point: &[Cell]) -> RunReport<RunMetrics> {
         let (bp, rx_eq, rx_ex) = composite_sma_cross_harness();
         let mut h = bp
             .bootstrap_with_cells(point)
@@ -945,7 +949,7 @@ mod tests {
     /// A fake report whose manifest `commit` encodes the input cells — enough to
     /// prove the sweep closure ran on exactly the given point, without a harness
     /// run. A free `fn` (Sync) so it serves directly as the `sweep` closure.
-    fn tagged_report(point: &[Cell]) -> RunReport {
+    fn tagged_report(point: &[Cell]) -> RunReport<RunMetrics> {
         RunReport {
             manifest: RunManifest {
                 commit: point.iter().map(|c| c.i64().to_string()).collect::<Vec<_>>().join(","),

crates/aura-engine/src/walkforward.rs

@@ -120,23 +120,23 @@ impl Iterator for WindowRoller {
 /// What a per-window run yields: the params chosen on the in-sample window, the
 /// recorded OOS pip-equity segment (for stitching), and the OOS [`RunReport`] (the
 /// per-window C18 record). Self-describing, analog to
-/// [`SweepPoint`](crate::SweepPoint)/[`McDraw`](crate::McDraw).
+/// [`SweepPoint`](crate::SweepPoint) (and the backtest layer's `McDraw`).
 #[derive(Clone, Debug, PartialEq)]
-pub struct WindowRun {
+pub struct WindowRun<M> {
     /// The chosen params as the tag-free coordinate [`Cell`]s the inner sweep
     /// produced (C7: the kind lives once on [`WalkForwardResult::space`], not at the
     /// value). The named/typed view is `zip_params(&result.space, &chosen_params)` —
     /// the same idiom as [`SweepFamily::named_params`](crate::SweepFamily).
     pub chosen_params: Vec<Cell>,
     pub oos_equity: Vec<(Timestamp, f64)>,
-    pub oos_report: RunReport,
+    pub oos_report: RunReport<M>,
 }
 
 /// One completed window: its bounds + the run the closure produced, in roll order.
 #[derive(Clone, Debug, PartialEq)]
-pub struct WindowOutcome {
+pub struct WindowOutcome<M> {
     pub bounds: WindowBounds,
-    pub run: WindowRun,
+    pub run: WindowRun<M>,
 }
 
 /// The result family of a walk-forward. Analog to [`SweepFamily`](crate::SweepFamily)
@@ -146,7 +146,7 @@ pub struct WindowOutcome {
 /// points and `optimize`/`rank_by` are computed on demand). Non-redundant by
 /// construction.
 #[derive(Clone, Debug, PartialEq)]
-pub struct WalkForwardResult {
+pub struct WalkForwardResult<M> {
     /// The param-space schema (kinds + slot names), once for the whole family —
     /// the kinds the per-window [`WindowRun::chosen_params`] cells are read against
     /// (C7). Identical for every window (same blueprint). Mirrors
@@ -154,7 +154,7 @@ pub struct WalkForwardResult {
     pub space: Vec<ParamSpec>,
     /// Per-window outcomes in roll order (window 0 first), independent of thread
     /// completion (C1).
-    pub windows: Vec<WindowOutcome>,
+    pub windows: Vec<WindowOutcome<M>>,
     /// The stitched continuous OOS pip-equity curve: each window's OOS segment
     /// offset by the running sum of all prior segments' final cumulative values,
     /// so equity carries forward across the IS/OOS boundaries (continuous, not a
@@ -165,7 +165,7 @@ pub struct WalkForwardResult {
     pub stitched_oos_equity: Vec<(Timestamp, f64)>,
 }
 
-impl WalkForwardResult {
+impl<M> WalkForwardResult<M> {
     /// The chosen params of the `window`-th outcome paired with their names — a
     /// derived view over the carried param-space (reuses [`zip_params`]); the
     /// walk-forward mirror of [`SweepFamily::named_params`](crate::SweepFamily).
@@ -179,16 +179,17 @@ impl WalkForwardResult {
 /// `run_indexed` (on whichever pool is ambient when called), stitch the OOS
 /// equity, and assemble the result. One source of truth so the two callers can
 /// never drift on assembly logic while their pool selection differs.
-fn assemble_walk_forward<F>(
+fn assemble_walk_forward<M, F>(
     space: Vec<ParamSpec>,
     bounds: Vec<WindowBounds>,
     run_window: &F,
-) -> WalkForwardResult
+) -> WalkForwardResult<M>
 where
-    F: Fn(WindowBounds) -> WindowRun + Sync,
+    M: Send,
+    F: Fn(WindowBounds) -> WindowRun<M> + Sync,
 {
     let runs = run_indexed(bounds.len(), |i| run_window(bounds[i]));
-    let windows: Vec<WindowOutcome> = bounds
+    let windows: Vec<WindowOutcome<M>> = bounds
         .into_iter()
         .zip(runs)
         .map(|(bounds, run)| WindowOutcome { bounds, run })
@@ -209,9 +210,10 @@ where
 /// stream `Vec` in this API. Precondition: the roller yields `>= 1` window
 /// (`WindowRoller::new` rejects an empty roll). Param stability is a separate
 /// on-demand reduction ([`param_stability`]), never computed here.
-pub fn walk_forward<F>(roller: WindowRoller, space: Vec<ParamSpec>, run_window: F) -> WalkForwardResult
+pub fn walk_forward<M, F>(roller: WindowRoller, space: Vec<ParamSpec>, run_window: F) -> WalkForwardResult<M>
 where
-    F: Fn(WindowBounds) -> WindowRun + Sync,
+    M: Send,
+    F: Fn(WindowBounds) -> WindowRun<M> + Sync,
 {
     let bounds: Vec<WindowBounds> = roller.collect();
     debug_assert!(!bounds.is_empty(), "walk_forward requires a non-empty roll");
@@ -227,14 +229,15 @@ where
 /// `walk_forward` (`F: Sync`): the install closure only borrows `run_window`,
 /// never moves it.
 #[cfg(test)]
-fn walk_forward_with_threads<F>(
+fn walk_forward_with_threads<M, F>(
     roller: WindowRoller,
     space: Vec<ParamSpec>,
     nthreads: usize,
     run_window: F,
-) -> WalkForwardResult
+) -> WalkForwardResult<M>
 where
-    F: Fn(WindowBounds) -> WindowRun + Sync,
+    M: Send,
+    F: Fn(WindowBounds) -> WindowRun<M> + Sync,
 {
     let bounds: Vec<WindowBounds> = roller.collect();
     debug_assert!(!bounds.is_empty(), "walk_forward requires a non-empty roll");
@@ -249,7 +252,7 @@ where
 /// segment's samples are offset by the running sum of all prior segments' final
 /// cumulative values, so equity carries forward across windows (no per-window
 /// reset). An empty segment adds `0.0` to the offset and no points.
-fn stitch(windows: &[WindowOutcome]) -> Vec<(Timestamp, f64)> {
+fn stitch<M>(windows: &[WindowOutcome<M>]) -> Vec<(Timestamp, f64)> {
     let mut out = Vec::new();
     let mut offset = 0.0_f64;
     for w in windows {
@@ -275,7 +278,7 @@ fn stitch(windows: &[WindowOutcome]) -> Vec<(Timestamp, f64)> {
 /// value, a `bool` slot to `0.0`/`1.0` (so `mean` is the fraction of windows that
 /// chose `true`). A `timestamp` slot is structurally impossible (C20: a timestamp
 /// is a structural axis, never a numeric knob).
-pub fn param_stability(result: &WalkForwardResult) -> Vec<MetricStats> {
+pub fn param_stability<M>(result: &WalkForwardResult<M>) -> Vec<MetricStats> {
     if result.windows.is_empty() {
         return Vec::new();
     }
@@ -305,9 +308,10 @@ pub fn param_stability(result: &WalkForwardResult) -> Vec<MetricStats> {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::{summarize, RunManifest};
+    use crate::RunManifest;
+    use aura_backtest::{summarize, RunMetrics};
 
-    fn dummy_report() -> RunReport {
+    fn dummy_report() -> RunReport<RunMetrics> {
         RunReport {
             manifest: RunManifest {
                 commit: "t".to_string(),
@@ -336,7 +340,7 @@ mod tests {
 
     /// A WindowOutcome with explicit OOS equity + chosen params (zero bounds — the
     /// stitch/stability reductions ignore bounds).
-    fn outcome(oos_equity: Vec<(Timestamp, f64)>, chosen: Vec<Cell>) -> WindowOutcome {
+    fn outcome(oos_equity: Vec<(Timestamp, f64)>, chosen: Vec<Cell>) -> WindowOutcome<RunMetrics> {
         WindowOutcome {
             bounds: WindowBounds {
                 is: (Timestamp(0), Timestamp(0)),
@@ -349,7 +353,7 @@ mod tests {
     /// A deterministic per-window run keyed by the OOS start, so distinct windows
     /// differ. No real harness — these tests pin the orchestration (roll order,
     /// stitch, stability); mc/sweep cover the harness path. Free `fn` => `Sync`.
-    fn run_window_fixture(w: WindowBounds) -> WindowRun {
+    fn run_window_fixture(w: WindowBounds) -> WindowRun<RunMetrics> {
         let Timestamp(k) = w.oos.0;
         WindowRun {
             chosen_params: vec![Cell::from_i64(k % 3), Cell::from_f64(k as f64 * 0.5)],

crates/aura-engine/tests/list_sweep_e2e.rs

@@ -11,10 +11,10 @@ use std::sync::mpsc;
 
 use aura_core::{Cell, Firing, ScalarKind, Timestamp};
 use aura_engine::{
-    f64_field, summarize, sweep, BlueprintNode, Composite, Edge, ListSpace, OutField, ParamSpec,
+    f64_field, sweep, BlueprintNode, Composite, Edge, ListSpace, OutField, ParamSpec,
     Role, RunManifest, RunReport, Scalar, SweepError, Target, VecSource,
 };
-use aura_backtest::SimBroker;
+use aura_backtest::{summarize, RunMetrics, SimBroker};
 use aura_std::{Recorder, Sma, Sub};
 use aura_strategy::Bias;
 
@@ -91,7 +91,7 @@ fn sma_cross_harness() -> (
 /// Build + bootstrap + run + summarize one point into a real `RunReport` — the
 /// full pipeline `ListSpace` must survive end-to-end, mirroring
 /// `random_sweep_e2e.rs::run_point`.
-fn run_point(point: &[Cell]) -> RunReport {
+fn run_point(point: &[Cell]) -> RunReport<RunMetrics> {
     let (bp, rx_eq, rx_ex) = sma_cross_harness();
     let mut h = bp.bootstrap_with_cells(point).expect("ListSpace validates points before sweep");
     h.run(vec![Box::new(VecSource::new(synthetic_prices()))]);

crates/aura-engine/tests/position_event_table.rs

@@ -1,11 +1,13 @@
 //! End-to-end verification of the C10 derived position-event table (#114) at the
 //! public crate boundary: the World consumes `PositionAction` / `PositionEvent`
-//! via the `aura_engine` re-exports and persists the audit table the same way
+//! via the `aura_backtest` re-exports (moved from `aura_engine`, #291) and
-//! the run registry persists a `RunReport` — through serde_json. These tests pin
+//! persists the audit table the same way the run registry persists a
-//! the *consumer-visible wire shape* (the persisted column form), distinct from
+//! `RunReport` — through serde_json. These tests pin the *consumer-visible
-//! the crate-internal type-level unit tests in `report.rs`.
+//! wire shape* (the persisted column form), distinct from the crate-internal
+//! type-level unit tests in `aura-backtest::metrics`.
 
-use aura_engine::{PositionAction, PositionEvent, Timestamp};
+use aura_backtest::{PositionAction, PositionEvent};
+use aura_engine::Timestamp;
 
 /// A minimal stop-and-reverse audit table as a broker would derive it: open
 /// long, then at one later instant Close-before-open into a short (a reversal,

crates/aura-engine/tests/r_sma_e2e.rs

@@ -26,8 +26,10 @@
 //!    `r_col_indices_match_producer_field_layout` test pins that contract.
 
 use aura_core::{AnyColumn, Ctx, Node, Scalar, ScalarKind, Timestamp};
-use aura_engine::{PositionAction, RMetrics, derive_position_events, summarize_r};
+use aura_backtest::{
-use aura_backtest::{PM_FIELD_NAMES, PM_RECORD_KINDS, PM_WIDTH, PositionManagement};
+    derive_position_events, summarize_r, PositionAction, PositionManagement, RMetrics,
+    PM_FIELD_NAMES, PM_RECORD_KINDS, PM_WIDTH,
+};
 use aura_strategy::{ConstantCost, CostSum, FixedStop, VolSlippageCost};
 
 // The indices `report.rs::summarize_r` reads the dense record by. Re-declared here

crates/aura-engine/tests/random_sweep_e2e.rs

@@ -15,10 +15,10 @@ use std::sync::mpsc;
 
 use aura_core::{Cell, Firing, ScalarKind, Timestamp};
 use aura_engine::{
-    f64_field, summarize, sweep, BlueprintNode, Composite, Edge, OutField, ParamRange, ParamSpec,
+    f64_field, sweep, BlueprintNode, Composite, Edge, OutField, ParamRange, ParamSpec,
     RandomSpace, Role, RunManifest, RunReport, Scalar, Space, SweepError, Target, VecSource,
 };
-use aura_backtest::SimBroker;
+use aura_backtest::{summarize, RunMetrics, SimBroker};
 use aura_std::{Recorder, Sma, Sub};
 use aura_strategy::Bias;
 
@@ -102,7 +102,7 @@ fn sma_cross_harness() -> (
 /// only the public surface. A fresh harness (fresh sink channels) per point keeps
 /// the runs disjoint (C1). The manifest is a fixed minimal fixture — only the
 /// metrics carry the run, so determinism makes this reproduce a point exactly.
-fn run_point(point: &[Cell]) -> RunReport {
+fn run_point(point: &[Cell]) -> RunReport<RunMetrics> {
     let (bp, rx_eq, rx_ex) = sma_cross_harness();
     let mut h = bp
         .bootstrap_with_cells(point)

crates/aura-engine/tests/streaming_reduction_equivalence.rs

@@ -5,8 +5,7 @@
 //! the full record, and a `SeriesReducer` summary equals `summarize`'s fields.
 
 use aura_core::{AnyColumn, Ctx, Node, Scalar, ScalarKind, Timestamp};
-use aura_engine::{summarize, summarize_r};
+use aura_backtest::{summarize, summarize_r, PM_RECORD_KINDS, PM_WIDTH};
-use aura_backtest::{PM_RECORD_KINDS, PM_WIDTH};
 use aura_std::{GatedRecorder, SeriesReducer};
 use std::sync::mpsc;
 

crates/aura-ingest/examples/ger40_breakout_compare.rs

@@ -24,7 +24,7 @@
 use std::sync::Arc;
 
 use aura_core::{Cell, Scalar, Timestamp};
-use aura_engine::{summarize, RunMetrics};
+use aura_backtest::{summarize, RunMetrics};
 use chrono_tz::Europe::Berlin;
 
 use aura_ingest::{default_data_server, open_ohlc, DataServer, DEFAULT_DATA_PATH};

crates/aura-ingest/examples/ger40_breakout_sweep.rs

@@ -16,7 +16,8 @@
 use std::sync::Arc;
 
 use aura_core::{Cell, Scalar, Timestamp};
-use aura_engine::{sweep, summarize, GridSpace, RunManifest, RunReport};
+use aura_backtest::{summarize, RunReport};
+use aura_engine::{sweep, GridSpace, RunManifest};
 use chrono_tz::Europe::Berlin;
 use aura_ingest::{default_data_server, open_ohlc, DataServer, DEFAULT_DATA_PATH};
 

crates/aura-ingest/examples/ger40_breakout_walkforward.rs

@@ -18,9 +18,10 @@
 use std::sync::Arc;
 
 use aura_core::{Cell, Scalar, Timestamp};
+use aura_backtest::{summarize, RunReport, WindowRun};
 use aura_engine::{
-    sweep, summarize, walk_forward, GridSpace, RollMode, RunManifest, RunReport, WindowBounds,
+    sweep, walk_forward, GridSpace, RollMode, RunManifest, WindowBounds,
-    WindowRoller, WindowRun,
+    WindowRoller,
 };
 use chrono::TimeZone;
 use chrono_tz::Europe::Berlin;

crates/aura-ingest/examples/shared/breakout_real.rs

@@ -40,10 +40,10 @@ use std::sync::mpsc;
 
 use aura_core::{NodeSchema, Scalar, ScalarKind, Timestamp};
 use aura_engine::{
-    join_on_ts, summarize, Composite, Edge, FlatGraph, GraphBuilder, Harness, RunManifest,
+    join_on_ts, Composite, Edge, FlatGraph, GraphBuilder, Harness, RunManifest,
-    RunReport, SourceSpec, Target,
+    SourceSpec, Target,
 };
-use aura_backtest::SimBroker;
+use aura_backtest::{summarize, RunReport, SimBroker};
 use aura_market::{Resample, Session};
 use aura_std::{And, Delay, EqConst, Gt, Latch, Recorder};
 use chrono::TimeZone;

crates/aura-ingest/tests/ger40_breakout_real.rs

@@ -31,7 +31,7 @@ fn run_real(
     server: &Arc<DataServer>,
     from: Timestamp,
     to: Timestamp,
-) -> (aura_engine::RunReport, Vec<(f64, f64)>) {
+) -> (aura_backtest::RunReport, Vec<(f64, f64)>) {
     let sources =
         open_ohlc(server, SYMBOL, from, to).expect("GER40 has data in the Sept-2024 window");
     let (mut h, taps) = build_harness();

crates/aura-ingest/tests/ger40_breakout_world.rs

@@ -21,9 +21,10 @@
 use std::sync::Arc;
 
 use aura_core::{Cell, Scalar, Timestamp};
+use aura_backtest::{summarize, RunReport, WindowRun};
 use aura_engine::{
-    sweep, summarize, walk_forward, GridSpace, RollMode, RunManifest, RunReport, WindowBounds,
+    sweep, walk_forward, GridSpace, RollMode, RunManifest, WindowBounds,
-    WindowRoller, WindowRun,
+    WindowRoller,
 };
 use chrono::TimeZone;
 use chrono_tz::Europe::Berlin;

crates/aura-ingest/tests/real_bars.rs

@@ -9,11 +9,11 @@ use std::sync::Arc;
 
 use aura_core::{Firing, NodeSchema, PortSpec, Scalar, ScalarKind, Timestamp};
 use aura_engine::{
-    f64_field, summarize, Edge, FlatGraph, Harness, RunManifest, RunReport, SourceSpec, Target,
+    f64_field, Edge, FlatGraph, Harness, RunManifest, SourceSpec, Target,
     VecSource,
 };
 use aura_ingest::load_m1_window;
-use aura_backtest::SimBroker;
+use aura_backtest::{summarize, RunReport, SimBroker};
 use aura_std::{Recorder, Sma, Sub};
 use aura_strategy::Bias;
 use data_server::{DataServer, DEFAULT_DATA_PATH};

crates/aura-ingest/tests/streaming_seam.rs

@@ -9,11 +9,11 @@ use std::sync::Arc;
 
 use aura_core::{Firing, NodeSchema, PortSpec, Scalar, ScalarKind, Timestamp};
 use aura_engine::{
-    f64_field, summarize, Edge, FlatGraph, Harness, RunManifest, RunReport, Source, SourceSpec,
+    f64_field, Edge, FlatGraph, Harness, RunManifest, Source, SourceSpec,
     Target,
 };
 use aura_ingest::{M1Field, M1FieldSource};
-use aura_backtest::SimBroker;
+use aura_backtest::{summarize, RunReport, SimBroker};
 use aura_std::{Recorder, Sma, Sub};
 use aura_strategy::Bias;
 use data_server::loader::CHUNK_SIZE;

crates/aura-registry/Cargo.toml

@@ -10,6 +10,15 @@ publish.workspace = true
 # back (admitted under the amended C16 per-case policy, INDEX.md). RunReport
 # derives serde in aura-engine.
 aura-engine = { path = "../aura-engine" }
+# aura-backtest carries the trading instantiation of the metric-generic engine
+# (M = RunMetrics: RunReport/SweepFamily/SweepPoint/McFamily/RBootstrap/
+# WalkForwardResult aliases, RMetrics, r_metrics_from_rs) the registry indexes
+# and ranks (#291/#292, C28 phase 2 — process column, C28-legal).
+aura-backtest = { path = "../aura-backtest" }
+# aura-analysis is the foundation statistics kernel (MetricStats/resample_block/
+# SplitMix64) `null_best_of_k`'s trials-deflation reality-check reduces through
+# (#291).
+aura-analysis = { path = "../aura-analysis" }
 # the document stores' content-id primitive (put_process/put_campaign key on
 # aura_research::content_id_of, the same hash the doc types canonicalize to);
 # also PrimitiveBuilder, the referential tier's resolver signature.

crates/aura-registry/src/compat.rs

@@ -10,9 +10,8 @@
 //! (`Registry::append`, `RunReport::to_json`) is untouched and still emits the
 //! tagged form, so this is a one-directional widening, not a format change.
 
-use aura_engine::{
+use aura_backtest::{RunMetrics, RunReport};
-    FamilySelection, ProjectProvenance, RunManifest, RunMetrics, RunReport, Scalar, Timestamp,
+use aura_engine::{FamilySelection, ProjectProvenance, RunManifest, Scalar, Timestamp};
-};
 use serde::de::{self, Deserializer};
 use serde::Deserialize;
 

crates/aura-registry/src/lib.rs

@@ -18,10 +18,11 @@ use std::io::Write;
 use std::path::{Path, PathBuf};
 
 use aura_core::PrimitiveBuilder;
+use aura_analysis::{resample_block, MetricStats, SplitMix64};
+use aura_backtest::{r_metrics_from_rs, RunReport, SweepFamily, SweepPoint};
 use aura_engine::{
-    blueprint_from_json, blueprint_identity_json, expected_max_of_normals, r_metrics_from_rs,
+    blueprint_from_json, blueprint_identity_json, expected_max_of_normals, FamilySelection,
-    resample_block, FamilySelection, MetricStats, RunReport, SelectionMode, SplitMix64,
+    SelectionMode,
-    SweepFamily, SweepPoint,
 };
 use aura_research::{CampaignDoc, DocRef};
 
@@ -528,7 +529,7 @@ fn resolve_metric(name: &str) -> Result<Metric, RegistryError> {
 /// block; a missing block reads `NEG_INFINITY` (the worst rank for the
 /// higher-is-better R keys), preserving `metric_cmp`'s documented contract.
 fn metric_value(rep: &RunReport, m: Metric) -> f64 {
-    fn r_get(rep: &RunReport, f: impl Fn(&aura_engine::RMetrics) -> f64) -> f64 {
+    fn r_get(rep: &RunReport, f: impl Fn(&aura_backtest::RMetrics) -> f64) -> f64 {
         rep.metrics.r.as_ref().map(f).unwrap_or(f64::NEG_INFINITY)
     }
     match m {
@@ -943,8 +944,8 @@ impl From<std::io::Error> for RegistryError {
 mod tests {
     use super::*;
     use aura_core::{Cell, Scalar, Timestamp};
-    use aura_engine::RMetrics;
+    use aura_backtest::{RMetrics, RunMetrics};
-    use aura_engine::{RunManifest, RunMetrics, RunReport, SweepFamily, SweepPoint};
+    use aura_engine::RunManifest;
 
     fn report_with(total_pips: f64, max_drawdown: f64, flips: u64) -> RunReport {
         RunReport {
@@ -1422,7 +1423,7 @@ mod tests {
         // `summarize_r`, which RETAINS it — and `optimize_deflated`'s R arm
         // resamples exactly that conduit, so a fixture without it cannot reach
         // the R-arm path under test.
-        let mut r = aura_engine::r_metrics_from_rs(&net_trade_rs);
+        let mut r = aura_backtest::r_metrics_from_rs(&net_trade_rs);
         r.net_trade_rs = net_trade_rs;
         SweepPoint {
             params: vec![],
@@ -2530,7 +2531,7 @@ mod tests {
 
     #[test]
     fn campaign_run_record_roundtrips_with_annotator_fields() {
-        use aura_engine::r_bootstrap;
+        use aura_backtest::r_bootstrap;
         let path = temp_family_dir("campaign_run_annotator_roundtrip");
         let reg = Registry::open(&path);
         // r_bootstrap is pure given its seed (C1), and serde_json round-trips

crates/aura-registry/src/lineage.rs

@@ -19,9 +19,8 @@ use std::fs;
 use std::io::Write;
 
 use aura_core::Scalar;
-use aura_engine::{
+use aura_backtest::{McFamily, RBootstrap, RunReport, SweepFamily, WalkForwardResult};
-    FamilySelection, McFamily, RBootstrap, RunReport, SweepFamily, WalkForwardResult,
+use aura_engine::FamilySelection;
-};
 use serde::{Deserialize, Serialize};
 
 use crate::{Generalization, Registry, RegistryError};

crates/aura-registry/tests/campaign_run_store_e2e.rs

@@ -12,7 +12,8 @@ use std::fs;
 use std::path::PathBuf;
 
 use aura_core::{Scalar, Timestamp};
-use aura_engine::{FamilySelection, RunManifest, RunMetrics, RunReport, SelectionMode};
+use aura_backtest::{RunMetrics, RunReport};
+use aura_engine::{FamilySelection, RunManifest, SelectionMode};
 use aura_registry::{
     CampaignRunRecord, CellRealization, FamilyKind, Registry, StageRealization, StageSelection,
 };

crates/aura-vocabulary/tests/c28_layering.rs

@@ -35,19 +35,15 @@ fn prod_intra_workspace_deps(crate_name: &str) -> Vec<String> {
 fn node_crates_obey_the_c28_import_direction() {
     // (crate, the intra-workspace [dependencies] C28 permits for its layer)
     let allowed: &[(&str, &[&str])] = &[
-        ("aura-std", &["aura-core"]),      // engine
+        ("aura-analysis", &[]),                          // foundation: no aura-* deps
-        ("aura-market", &["aura-core"]),   // market
+        ("aura-std", &["aura-core"]),
-        ("aura-strategy", &["aura-core"]), // strategy
+        ("aura-market", &["aura-core"]),
-        ("aura-backtest", &["aura-core"]), // backtest (may gain aura-strategy when it imports one)
+        ("aura-strategy", &["aura-core"]),
+        ("aura-engine", &["aura-core", "aura-analysis"]),
+        ("aura-backtest", &["aura-core", "aura-engine"]),
         (
             "aura-vocabulary",
-            &[
+            &["aura-core", "aura-std", "aura-market", "aura-strategy", "aura-backtest"],
-                "aura-core",
-                "aura-std",
-                "aura-market",
-                "aura-strategy",
-                "aura-backtest",
-            ],
         ),
     ];
     for (crate_name, permitted) in allowed {

docs/design/INDEX.md

@@ -2666,26 +2666,27 @@ measurement and backtest instead of baked in R-only — this is #147).
 shell", #286).** This contract states the *target*. At HEAD the crates cut by
 mechanical role (vocabulary / reductions / documents / orchestration / CLI), not
 cleanly by layer, so the layering is realized only partially:
-- The dependency *direction* already nearly obeys the rule. The one hard
+- The dependency *direction* now obeys the rule across the engine stack. The
-  production violation is the engine's re-export of the backtest metrics —
+  `aura-engine → aura-backtest` production violation is cut (phase 2, #292):
-  since the #291 split, `aura-engine → aura-backtest`: `report.rs` re-exports
+  `RunReport` is generic over its metric payload `M` (the engine names no
-  the R-metrics and the position-event table (`RMetrics`/`RunMetrics`/
+  concrete metric type), the pip/R reductions and the MC assembly (`summarize`,
-  `summarize_r`/`r_metrics_from_rs`, `PositionEvent`/`PositionAction`/
+  `McAggregate`/`RBootstrap`/`monte_carlo`) moved to `aura-backtest`, and the
-  `derive_position_events`), and the generic-statistics kernel in `mc.rs`
+  statistics kernel (`MetricStats`/`quantile`/`resample_block`/`SplitMix64`)
-  carries R/bias-typed fields (`McAggregate.bias_sign_flips`, `RBootstrap.e_r`)
+  moved to the `aura-analysis` foundation. The engine's `[dependencies]` are now
-  inside otherwise domain-free structs. The `aura-engine → aura-analysis` edge
+  `aura-core` + `aura-analysis` only; the backtest layer instantiates
-  is no longer a violation: post-#291 that crate holds only domain-free
+  `M = RunMetrics` (`aura-backtest → aura-engine`, an outer→inner edge). The
-  statistics + selection provenance (`FamilySelection` for
+  ladder direction is enforced for the engine/backtest/analysis rungs by the
-  `RunManifest.selection`), foundation-grade under this contract.
+  structural test (`c28_layering`, seven-row table).
-- **That violation is the exact surface of the deliberately-deferred #147.**
+- **#147 split, partially retired (user-ratified 2026-07-20).** Item 1 — the
-  Cutting it cleanly needs the metric genericity (`RunReport<M>` plus abstracting
+  metric genericity `RunReport<M>` — shipped as phase 2 above (#292). Item 2 —
-  the registry deflation vocabulary) that #136 ratified *deferring* (user-ratified
+  abstracting the registry `Metric`/`metric_cmp`/deflation vocabulary — stays
-  2026-06-27: the registry `Metric`/`metric_cmp`/deflation vocabulary "stays in
+  deferred: the registry sits in the process column, whose trading-awareness
-  the registry by design … forcing genericity would be a one-implementor
+  this contract permits (`aura-registry → aura-research`, column-internal and
-  abstraction", deferred to the World/C21 layer — see the C16 realization note).
+  legal), and the tree still holds exactly one metric consumer — a measurement
-  The stratification supplies the *second* metric consumer (measurement) that the
+  run reports tap names, not metrics. It lifts when measurement supplies its
-  deferral was waiting on, so retiring #147 and cutting this edge are one coupled
+  first deflatable metric (a tap-to-scalar reduction carrying its own null
-  decision — reserved to the user, since it reverses a ratified deferral.
+  model, #290); the original one-implementor rationale (#136, user-ratified
+  2026-06-27) holds until then. See #147 for the full disposition.
 - The `aura-std` four-layer roster is now cut by layer (phase 4, #288): `aura-std`
   holds the engine nodes only (arithmetic/logic/rolling + the generic sinks);
   `aura-market` (`session`, `resample`), `aura-strategy` (`bias`/`stop_rule`/
@@ -2701,7 +2702,11 @@ cleanly by layer, so the layering is realized only partially:
   column-internal and legal.
 - Phased realization (each independently shippable; behaviour byte-identical
   except the purely additive shape dispatch): (1) this contract; (2) cut the
-  `engine → analysis` edge — coupled to the #147 decision above; (3) the #286
+  engine's backtest-metrics edge via `RunReport<M>` — **done** (#292:
+  metric-generic run record, the pip/R reductions + MC assembly relocated to
+  `aura-backtest`, the statistics kernel to `aura-analysis`, dependency
+  inversion, ladder enforced by the structural test; realizes item 1 of #147);
+  (3) the #286
   shape dispatch plus the per-run scaffold as a library (seeds an `aura-backtest`
   crate, gives *measurement* its run verb, removes the measured O(cycles) run-time
   retention); (4) split the `aura-std` roster along engine / market / strategy /