feat(0092): run-from-blueprint infrastructure — restructure + topology_hash + shared seam (Tasks 1-3)

Cycle-1 infrastructure for #165 (World/C21). Behaviour-preserving + workspace
green; the CLI arm + tests (Tasks 4-5) follow.

Task 1 — restructured stage1_r_graph into stage1_signal() (the serializable
signal leg: SMA-cross -> Bias, a `price` input-role + a `bias` output) +
wrap_stage1r(signal, ...) (broker/sinks/exec/cost around a nested signal), with
stage1_r_graph() = wrap_stage1r(stage1_signal(...)). DEVIATION from the plan's
"callers untouched" claim, verified behaviour-preserving: nesting the signal
prefixes its param-space names (stage1_signal.fast.length), which broke the
sweep's bare .axis("fast.length"); fixed exactly as the #137 stop-knob machinery
does — FAST_LENGTH_SUFFIX/SLOW_LENGTH_SUFFIX suffix-resolution + stage1_r_friendly_name
arms mapping the prefixed names back to the bare manifest names. All observable
behaviour (manifest names, member keys, metrics, traces) byte-identical; the flat
graph and every metric unchanged (full suite green, incl. sweep/walkforward/MC).

Task 2 — RunManifest.topology_hash: Option<String> (Tier-1 optional, serde
default/skip — old manifests byte-identical, #156), threaded None into all 24
literal sites across 7 crates. Also fixed the aura-registry RunManifestRead
read-mirror to carry topology_hash (it was hardcoding None on load) — else a
stored hash would silently drop on the registry round-trip once Task 3 stores a
real one.

Task 3 — the shared run_signal_stage1r seam (hash the signal, wrap, compile with
params, bootstrap, run, manifest with params from param_space + topology_hash) +
the sha256_hex topology_hash helper (sha2 in aura-cli, off the frozen engine,
invariant 8). run_stage1_r now carries its topology_hash too (every run becomes
self-identifying, #158). The seam is unwired until Task 4 (transient
#[allow(dead_code)], retired there); RED-first seam tests added.

Verified: cargo test --workspace green (51 suites, 0 failures); cargo clippy
--workspace --all-targets -D warnings clean.

refs #165

Cargo.lock

@@ -69,6 +69,7 @@ dependencies = [
  "libc",
  "serde",
  "serde_json",
+ "sha2",
 ]
 
 [[package]]
@@ -735,6 +736,17 @@ dependencies = [
  "digest",
 ]
 
+[[package]]
+name = "sha2"
+version = "0.10.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a7507d819769d01a365ab707794a4084392c824f54a7a6a7862f8c3d0892b283"
+dependencies = [
+ "cfg-if",
+ "cpufeatures",
+ "digest",
+]
+
 [[package]]
 name = "shlex"
 version = "2.0.1"

crates/aura-cli/Cargo.toml

@@ -35,3 +35,8 @@ serde_json = { workspace = true }
 # EPIPE. The standard vetted crate for signal disposition, already transitive in
 # Cargo.lock; admitted under the per-case dependency policy.
 libc = "0.2"
+# sha2: the run's topology_hash (#158) is the SHA256 of the canonical signal
+# blueprint serialization. A vetted RustCrypto crate (per-case C16 review,
+# SHA256 user-settled); lives in the research-side CLI, off the frozen engine
+# (invariant 8).
+sha2 = "0.10"

crates/aura-cli/src/main.rs

@@ -18,11 +18,12 @@ use render::{ChartData, ChartMeta, ChartMode, ReduceKind, Series};
 use aura_core::{zip_params, Cell, Firing, ParamSpec, Scalar, ScalarKind, Timestamp};
 use aura_composites::{cost_graph, risk_executor, risk_executor_vol_open, StopRule};
 use aura_engine::{
-    f64_field, join_on_ts, monte_carlo, param_stability, r_bootstrap, r_metrics_from_rs, summarize,
-    summarize_r, walk_forward, window_of, ColumnarTrace, Composite, Edge, FamilySelection, FlatGraph,
-    GraphBuilder, Harness, JoinedRow, McAggregate, McFamily, RBootstrap, RollMode, RunManifest,
-    RunMetrics, RunReport, SelectionMode, SourceSpec, SweepFamily, SweepPoint, SyntheticSpec, Target,
-    VecSource, WalkForwardResult, WindowBounds, WindowRoller, WindowRun,
+    blueprint_from_json, blueprint_to_json, f64_field, join_on_ts, monte_carlo, param_stability, r_bootstrap,
+    r_metrics_from_rs, summarize, summarize_r, walk_forward, window_of, BlueprintNode, ColumnarTrace,
+    Composite, Edge, FamilySelection, FlatGraph, GraphBuilder, Harness, JoinedRow, McAggregate,
+    McFamily, RBootstrap, RollMode, RunManifest, RunMetrics, RunReport, SelectionMode, SourceSpec,
+    SweepFamily, SweepPoint, SyntheticSpec, Target, VecSource, WalkForwardResult,
+    WindowBounds, WindowRoller, WindowRun,
 };
 use aura_registry::{
     check_r_metric, generalization, group_families, mc_member_reports, optimize_deflated,
@@ -32,7 +33,7 @@ use aura_registry::{
 use aura_std::{
     Add, Bias, CarryCost, ConstantCost, Delay, Ema, GatedRecorder, Gt, Latch, LinComb, LongOnly, Mul,
     Recorder, RollingMax, RollingMin, SeriesReducer, SimBroker, Sma, Sqrt, Sub, VolSlippageCost,
-    PM_FIELD_NAMES, PM_RECORD_KINDS,
+    std_vocabulary, PM_FIELD_NAMES, PM_RECORD_KINDS,
 };
 use std::sync::mpsc::{self, Receiver};
 use std::sync::LazyLock;
@@ -177,6 +178,7 @@ fn sim_optimal_manifest(
         broker: format!("sim-optimal(pip_size={pip_size})"),
         selection: None,
         instrument: None,
+        topology_hash: None,
     }
 }
 
@@ -1173,6 +1175,15 @@ const STOP_LENGTH_SUFFIX: &str = ".vol_stop.stop_length.length";
 /// (the `LinComb` named `stop_k`, whose single weight is `weights[0]`).
 const STOP_K_SUFFIX: &str = ".vol_stop.stop_k.weights[0]";
 
+/// The path-qualified `param_space()` suffixes of the two open SMA-cross signal knobs.
+/// Since cycle 0092 the signal leg is a nested `stage1_signal` composite, so its knobs
+/// land in `param_space` under the composite prefix (`stage1_signal.fast.length` /
+/// `stage1_signal.slow.length`); like the stop knobs they are resolved by suffix so the
+/// prefix is never hand-synced, and rendered back to the bare `fast.length` / `slow.length`
+/// manifest names by `stage1_r_friendly_name` (the pre-0092 family-record names — C18).
+const FAST_LENGTH_SUFFIX: &str = ".fast.length";
+const SLOW_LENGTH_SUFFIX: &str = ".slow.length";
+
 /// The friendly manifest name for an open vol-stop slot, given its path-qualified
 /// `param_space()` name: the two stop knobs render as `stop_length` / `stop_k` (their
 /// pre-#137 manual manifest names), every other slot unchanged. Decoupling the manifest
@@ -1183,6 +1194,10 @@ fn stage1_r_friendly_name(space_name: &str) -> String {
         "stop_length".to_string()
     } else if space_name.ends_with(STOP_K_SUFFIX) {
         "stop_k".to_string()
+    } else if space_name.ends_with(FAST_LENGTH_SUFFIX) {
+        "fast.length".to_string()
+    } else if space_name.ends_with(SLOW_LENGTH_SUFFIX) {
+        "slow.length".to_string()
     } else {
         space_name.to_string()
     }
@@ -1215,9 +1230,19 @@ fn stage1_r_sweep_family(trace: Option<&str>, data: &DataSource, grid: &Stage1RG
         .map(|p| p.name.clone())
         .find(|n| n.ends_with(STOP_K_SUFFIX))
         .expect("open stage1-r vol-stop exposes a stop_k axis");
+    let fast_axis = space
+        .iter()
+        .map(|p| p.name.clone())
+        .find(|n| n.ends_with(FAST_LENGTH_SUFFIX))
+        .expect("open stage1-r signal exposes a fast.length axis");
+    let slow_axis = space
+        .iter()
+        .map(|p| p.name.clone())
+        .find(|n| n.ends_with(SLOW_LENGTH_SUFFIX))
+        .expect("open stage1-r signal exposes a slow.length axis");
     let binder = bp
-        .axis("fast.length", grid.fast.clone())
-        .axis("slow.length", grid.slow.clone())
+        .axis(&fast_axis, grid.fast.clone())
+        .axis(&slow_axis, grid.slow.clone())
         .axis(&stop_length_axis, grid.stop_length.clone())
         .axis(&stop_k_axis, grid.stop_k.clone());
     // the varying axes drive the member key; translate the deep stop-axis names to the
@@ -1309,13 +1334,28 @@ fn stage1_r_sweep_over(from: Timestamp, to: Timestamp, data: &DataSource, grid:
     let (tx_req, _) = mpsc::channel();
     let bp = stage1_r_graph(tx_eq, tx_ex, tx_r, tx_req, None, None, true, true, None);
     let space = bp.param_space();
-    let stop_length_axis = space.iter().map(|p| p.name.clone())
-        .find(|n| n.ends_with(STOP_LENGTH_SUFFIX)).expect("open stage1-r vol-stop exposes a stop_length axis");
-    let stop_k_axis = space.iter().map(|p| p.name.clone())
-        .find(|n| n.ends_with(STOP_K_SUFFIX)).expect("open stage1-r vol-stop exposes a stop_k axis");
-    bp
-        .axis("fast.length", grid.fast.clone())
-        .axis("slow.length", grid.slow.clone())
+    let stop_length_axis = space
+        .iter()
+        .map(|p| p.name.clone())
+        .find(|n| n.ends_with(STOP_LENGTH_SUFFIX))
+        .expect("open stage1-r vol-stop exposes a stop_length axis");
+    let stop_k_axis = space
+        .iter()
+        .map(|p| p.name.clone())
+        .find(|n| n.ends_with(STOP_K_SUFFIX))
+        .expect("open stage1-r vol-stop exposes a stop_k axis");
+    let fast_axis = space
+        .iter()
+        .map(|p| p.name.clone())
+        .find(|n| n.ends_with(FAST_LENGTH_SUFFIX))
+        .expect("open stage1-r signal exposes a fast.length axis");
+    let slow_axis = space
+        .iter()
+        .map(|p| p.name.clone())
+        .find(|n| n.ends_with(SLOW_LENGTH_SUFFIX))
+        .expect("open stage1-r signal exposes a slow.length axis");
+    bp.axis(&fast_axis, grid.fast.clone())
+        .axis(&slow_axis, grid.slow.clone())
         .axis(&stop_length_axis, grid.stop_length.clone())
         .axis(&stop_k_axis, grid.stop_k.clone())
         .sweep_with_lattice(|point| {
@@ -2566,6 +2606,19 @@ const STAGE1_R_STOP_LENGTH: i64 = 3;
 /// `StopRule::Vol` and its manifest record, like [`STAGE1_R_STOP_LENGTH`].
 const STAGE1_R_STOP_K: f64 = 2.0;
 
+/// The stage1-r demo signal SMA lengths (fast/slow), bound at single-run build time.
+/// Single source for the graph that runs (`stage1_r_graph`), the signal that is hashed
+/// (`stage1_signal` → `topology_hash`), and the recorded manifest params — so the
+/// hashed topology cannot drift from the executed one across the function boundary.
+const STAGE1_R_SMA_FAST: i64 = 2;
+const STAGE1_R_SMA_SLOW: i64 = 4;
+
+/// The stage1-r demo `Bias` scale (conviction magnitude). Single source for the
+/// `Bias` node bound in `stage1_signal` (the hashed + executed signal) and the
+/// recorded manifest param, so the hashed topology cannot drift from the recorded
+/// `bias_scale` across the function boundary — the same guard as the SMA lengths.
+const STAGE1_R_BIAS_SCALE: f64 = 0.5;
+
 /// Short-horizon realized-range window for vol-scaled slippage. Deliberately
 /// distinct from `STAGE1_R_STOP_LENGTH` (3): scaling slippage by the stop's own
 /// vol would collapse cost-in-R to a constant (spec 0082). Short enough to warm
@@ -2609,10 +2662,55 @@ fn stage1_r_prices() -> Vec<(Timestamp, Scalar)> {
 static COL_PORTS: LazyLock<Vec<String>> =
     LazyLock::new(|| (0..PM_FIELD_NAMES.len()).map(|i| format!("col[{i}]")).collect());
 
+/// The Stage-1 signal leg as a standalone, serializable Composite: SMA-cross →
+/// Bias. Exposes a `price` input-role and a single `bias` output — the boundary
+/// shape a serialized blueprint round-trips (the `blueprint_serde_e2e.rs`
+/// template). The two SMA knobs are bound when `Some` (byte-identical to the old
+/// build-time bind) and left open as `fast.length` / `slow.length` when `None`.
+fn stage1_signal(fast_len: Option<i64>, slow_len: Option<i64>) -> Composite {
+    let mut g = GraphBuilder::new("stage1_signal");
+    let mut fast_b = Sma::builder().named("fast");
+    if let Some(l) = fast_len {
+        fast_b = fast_b.bind("length", Scalar::i64(l));
+    }
+    let fast = g.add(fast_b);
+    let mut slow_b = Sma::builder().named("slow");
+    if let Some(l) = slow_len {
+        slow_b = slow_b.bind("length", Scalar::i64(l));
+    }
+    let slow = g.add(slow_b);
+    let spread = g.add(Sub::builder());
+    let exposure = g.add(
+        Bias::builder()
+            .named("bias")
+            .bind("scale", Scalar::f64(STAGE1_R_BIAS_SCALE)),
+    );
+    let price = g.source_role("price", ScalarKind::F64);
+    g.feed(price, vec![fast.input("series"), slow.input("series")]);
+    g.connect(fast.output("value"), spread.input("lhs"));
+    g.connect(slow.output("value"), spread.input("rhs"));
+    g.connect(spread.output("value"), exposure.input("signal"));
+    g.expose(exposure.output("bias"), "bias");
+    g.build().expect("stage1 signal wiring resolves")
+}
+
+/// SHA256 (hex) of the canonical (#164, no-trailing-newline) serialization of a
+/// signal blueprint — the run's `topology_hash` (#158). Research-side (aura-cli),
+/// off the frozen engine (invariant 8).
+fn topology_hash(signal: &Composite) -> String {
+    use sha2::{Digest, Sha256};
+    let canonical = blueprint_to_json(signal).expect("a buildable signal serializes");
+    let digest = Sha256::digest(canonical.as_bytes());
+    digest.iter().map(|b| format!("{b:02x}")).collect()
+}
+
 /// The Stage-1 R harness topology, shared by the single run and the sweep. The two
 /// signal knobs are bound when `Some` (single run — identical to the old
 /// build-time bind, output byte-unchanged) and left free when `None` (sweep — they
-/// land in `param_space` as `fast.length` / `slow.length`). The vol-stop knobs are
+/// now nest in the `stage1_signal` composite, so they land in `param_space` as
+/// `stage1_signal.fast.length` / `stage1_signal.slow.length`, rendered back to the
+/// bare `fast.length` / `slow.length` manifest names by `stage1_r_friendly_name`).
+/// The vol-stop knobs are
 /// bound to the pinned constants when `stop_open` is `false` (single run + the
 /// default sweep, output byte-unchanged) and left free when `true` (a gridded sweep
 /// — they land in `param_space` under the `.vol_stop.stop_length.length` /
@@ -2638,21 +2736,43 @@ fn stage1_r_graph(
     stop_open: bool,
     reduce: bool,
     cost: Option<(CostConfig, mpsc::Sender<(Timestamp, Vec<Scalar>)>, mpsc::Sender<(Timestamp, Vec<Scalar>)>)>,
+) -> Composite {
+    wrap_stage1r(
+        stage1_signal(fast_len, slow_len),
+        tx_eq,
+        tx_ex,
+        tx_r,
+        tx_req,
+        stop_open,
+        reduce,
+        cost,
+    )
+}
+
+/// Wrap a Stage-1 `signal` composite (a `price`→`bias` leg) in the Stage-1 R run
+/// scaffolding: pip broker, the per-tap recorders, the vol-stop RiskExecutor, the
+/// r_equity / cost legs. The signal is nested and its `price`/`bias` boundary is
+/// wired across; everything else is verbatim from the old `stage1_r_graph` body, so
+/// a serialized signal loaded via `blueprint_from_json` runs through exactly the
+/// scaffolding the Rust-built signal does.
+#[allow(clippy::type_complexity, clippy::too_many_arguments)]
+fn wrap_stage1r(
+    signal: Composite,
+    tx_eq: mpsc::Sender<(Timestamp, Vec<Scalar>)>,
+    tx_ex: mpsc::Sender<(Timestamp, Vec<Scalar>)>,
+    tx_r: mpsc::Sender<(Timestamp, Vec<Scalar>)>,
+    tx_req: mpsc::Sender<(Timestamp, Vec<Scalar>)>,
+    stop_open: bool,
+    reduce: bool,
+    cost: Option<(
+        CostConfig,
+        mpsc::Sender<(Timestamp, Vec<Scalar>)>,
+        mpsc::Sender<(Timestamp, Vec<Scalar>)>,
+    )>,
 ) -> Composite {
     let mut g = GraphBuilder::new("stage1_r");
-    // SMA-cross signal → Bias (the same signal the pip sample uses).
-    let mut fast_b = Sma::builder().named("fast");
-    if let Some(l) = fast_len {
-        fast_b = fast_b.bind("length", Scalar::i64(l));
-    }
-    let fast = g.add(fast_b);
-    let mut slow_b = Sma::builder().named("slow");
-    if let Some(l) = slow_len {
-        slow_b = slow_b.bind("length", Scalar::i64(l));
-    }
-    let slow = g.add(slow_b);
-    let spread = g.add(Sub::builder());
-    let exposure = g.add(Bias::builder().named("bias").bind("scale", Scalar::f64(0.5)));
+    // SMA-cross signal → Bias, nested as a serializable `price`→`bias` leg.
+    let sig = g.add(BlueprintNode::Composite(signal));
     // pip branch (verbatim from sample_blueprint_with_sinks).
     let broker = g.add(SimBroker::builder(SYNTHETIC_PIP_SIZE));
     // R branch: bias + price → RiskExecutor(vol_stop) → dense R-record. The stop knobs
@@ -2701,8 +2821,7 @@ fn stage1_r_graph(
     };
     let price = g.source_role("price", ScalarKind::F64);
     let mut price_targets = vec![
-        fast.input("series"),
-        slow.input("series"),
+        sig.input("price"),
         broker.input("price"),
         exec.input("price"),
     ];
@@ -2711,12 +2830,9 @@ fn stage1_r_graph(
         price_targets.push(vlo.input("series"));
     }
     g.feed(price, price_targets);
-    g.connect(fast.output("value"), spread.input("lhs"));
-    g.connect(slow.output("value"), spread.input("rhs"));
-    g.connect(spread.output("value"), exposure.input("signal"));
-    g.connect(exposure.output("bias"), broker.input("exposure"));
-    g.connect(exposure.output("bias"), ex.input("col[0]"));
-    g.connect(exposure.output("bias"), exec.input("bias"));
+    g.connect(sig.output("bias"), broker.input("exposure"));
+    g.connect(sig.output("bias"), ex.input("col[0]"));
+    g.connect(sig.output("bias"), exec.input("bias"));
     g.connect(broker.output("equity"), eq.input("col[0]"));
     for (i, field) in PM_FIELD_NAMES.iter().enumerate() {
         g.connect(exec.output(field), rrec.input(COL_PORTS[i].as_str()));
@@ -2789,6 +2905,71 @@ fn stage1_r_graph(
     g.build().expect("stage1_r wiring resolves")
 }
 
+/// Resolve a `RunData` selector to the `(sources, window, pip_size)` triple the
+/// stage1-r run paths feed to the harness: the built-in synthetic Stage-1-R stream,
+/// or a lazily-streamed real M1 close source (with its sidecar pip + probed window).
+/// One definition shared by `run_stage1_r` and `run_signal_stage1r` so the
+/// source/window/pip wiring cannot drift between the two paths.
+#[allow(clippy::type_complexity)]
+fn resolve_run_data(
+    data: &RunData,
+) -> (
+    Vec<Box<dyn aura_engine::Source>>,
+    (Timestamp, Timestamp),
+    f64,
+) {
+    match data {
+        RunData::Synthetic => {
+            let sources: Vec<Box<dyn aura_engine::Source>> =
+                vec![Box::new(VecSource::new(stage1_r_prices()))];
+            let window = window_of(&sources).expect("non-empty synthetic stream");
+            (sources, window, SYNTHETIC_PIP_SIZE)
+        }
+        RunData::Real { symbol, from, to } => {
+            let (source, window, pip_size) = open_real_source(symbol, *from, *to);
+            (vec![source], window, pip_size)
+        }
+    }
+}
+
+/// Run a signal blueprint through the Stage-1-R scaffolding: hash the signal,
+/// wrap it (broker + equity/exposure/R sinks), compile with `params`, bootstrap,
+/// run over `data`, and build the RunReport (manifest carries topology_hash).
+/// The single construction+run path shared by the `aura run <blueprint.json>` CLI
+/// arm and its bit-identical test.
+fn run_signal_stage1r(signal: Composite, params: &[Scalar], data: RunData, seed: u64) -> RunReport {
+    let topo = topology_hash(&signal); // before signal is consumed
+    let names: Vec<String> = signal
+        .param_space()
+        .iter()
+        .map(|p| p.name.clone())
+        .collect();
+    let (tx_eq, rx_eq) = mpsc::channel();
+    let (tx_ex, rx_ex) = mpsc::channel();
+    let (tx_r, rx_r) = mpsc::channel();
+    // The req tap (r_equity recorder) is wired but not persisted on this path; keep the
+    // receiver alive so the sink's sends do not fail, but do not drain it.
+    let (tx_req, _rx_req) = mpsc::channel();
+    let wrapped = wrap_stage1r(signal, tx_eq, tx_ex, tx_r, tx_req, false, false, None);
+    let flat = wrapped
+        .compile_with_params(params)
+        .expect("signal binds + wraps to a valid harness");
+    let mut h = Harness::bootstrap(flat).expect("valid stage1-r harness");
+    let (sources, window, pip_size) = resolve_run_data(&data);
+    h.run(sources);
+    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 r_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_r.try_iter().collect();
+    let named_params: Vec<(String, Scalar)> =
+        names.into_iter().zip(params.iter().copied()).collect();
+    let mut manifest = sim_optimal_manifest(named_params, window, seed, pip_size);
+    manifest.broker = stage1_r_broker_label(pip_size);
+    manifest.topology_hash = Some(topo);
+    let mut metrics = summarize(&f64_field(&eq_rows, 0), &f64_field(&ex_rows, 0));
+    metrics.r = Some(summarize_r(&r_rows, &[]));
+    RunReport { manifest, metrics }
+}
+
 /// The stage1-r harness with its signal leg swapped for a Donchian channel breakout:
 /// `close -> Delay(1) -> {RollingMax,RollingMin}(channel) -> {Gt,Gt} -> {Latch,Latch}
 /// -> Sub = bias in {-1,0,+1}`. The one Delay(1) on close feeds both rolling nodes, so
@@ -3027,28 +3208,36 @@ fn run_stage1_r(
     let (tx_req, rx_req) = mpsc::channel();
     let (tx_net, rx_net) = mpsc::channel();
     let (tx_cost, rx_cost) = mpsc::channel();
-    let cost_bundle = if const_cost.is_some() || slip_vol_mult.is_some() || carry_per_cycle.is_some() {
-        Some((CostConfig { const_cost, slip_vol_mult, carry_per_cycle }, tx_net, tx_cost))
-    } else {
-        None
-    };
-    let flat = stage1_r_graph(tx_eq, tx_ex, tx_r, tx_req, Some(2), Some(4), false, false, cost_bundle)
-        .compile_with_params(&[])
-        .expect("valid stage1-r blueprint");
+    let cost_bundle =
+        if const_cost.is_some() || slip_vol_mult.is_some() || carry_per_cycle.is_some() {
+            Some((
+                CostConfig {
+                    const_cost,
+                    slip_vol_mult,
+                    carry_per_cycle,
+                },
+                tx_net,
+                tx_cost,
+            ))
+        } else {
+            None
+        };
+    let flat = stage1_r_graph(
+        tx_eq,
+        tx_ex,
+        tx_r,
+        tx_req,
+        Some(STAGE1_R_SMA_FAST),
+        Some(STAGE1_R_SMA_SLOW),
+        false,
+        false,
+        cost_bundle,
+    )
+    .compile_with_params(&[])
+    .expect("valid stage1-r blueprint");
     let mut h = Harness::bootstrap(flat).expect("valid stage1-r harness");
 
-    let (sources, window, pip_size): (Vec<Box<dyn aura_engine::Source>>, _, f64) = match &data {
-        RunData::Synthetic => {
-            let sources: Vec<Box<dyn aura_engine::Source>> =
-                vec![Box::new(VecSource::new(stage1_r_prices()))];
-            let window = window_of(&sources).expect("non-empty synthetic stream");
-            (sources, window, SYNTHETIC_PIP_SIZE)
-        }
-        RunData::Real { symbol, from, to } => {
-            let (source, window, pip_size) = open_real_source(symbol, *from, *to);
-            (vec![source], window, pip_size)
-        }
-    };
+    let (sources, window, pip_size) = resolve_run_data(&data);
     h.run(sources);
 
     let eq_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_eq.try_iter().collect();
@@ -3063,9 +3252,9 @@ fn run_stage1_r(
 
     let mut manifest = sim_optimal_manifest(
         vec![
-            ("sma_fast".to_string(), Scalar::i64(2)),
-            ("sma_slow".to_string(), Scalar::i64(4)),
-            ("bias_scale".to_string(), Scalar::f64(0.5)),
+            ("sma_fast".to_string(), Scalar::i64(STAGE1_R_SMA_FAST)),
+            ("sma_slow".to_string(), Scalar::i64(STAGE1_R_SMA_SLOW)),
+            ("bias_scale".to_string(), Scalar::f64(STAGE1_R_BIAS_SCALE)),
             ("stop_length".to_string(), Scalar::i64(STAGE1_R_STOP_LENGTH)), // vol_stop EWMA length
             ("stop_k".to_string(), Scalar::f64(STAGE1_R_STOP_K)),           // vol_stop multiplier (1R = k·σ)
         ],
@@ -3074,6 +3263,10 @@ fn run_stage1_r(
         pip_size,
     );
     manifest.broker = stage1_r_broker_label(pip_size);
+    manifest.topology_hash = Some(topology_hash(&stage1_signal(
+        Some(STAGE1_R_SMA_FAST),
+        Some(STAGE1_R_SMA_SLOW),
+    )));
     if let Some(name) = trace {
         persist_traces_r(name, &manifest, &eq_rows, &ex_rows, &req_rows, &net_rows);
     }
@@ -3241,6 +3434,88 @@ fn parse_run_args(rest: &[&str]) -> Result<RunArgs, String> {
     Ok(RunArgs { harness, data, trace, cost, slip_vol_mult, carry_per_cycle })
 }
 
+/// The parsed tail of `aura run <blueprint.json> [flags]`: the typed param cells bound
+/// at `compile_with_params` (in `param_space` order), the data source, and the manifest
+/// seed. The blueprint file itself is the topology selector, so there is no `--harness`
+/// on this path (the `.json` IS the harness). Pure — `main` reads the file, loads the
+/// blueprint, and runs.
+#[derive(Debug)]
+struct BlueprintRunArgs {
+    params: Vec<Scalar>,
+    data: RunData,
+    seed: u64,
+}
+
+/// Parse the `--params` value: a JSON array of externally-tagged `Scalar` cells
+/// (`[{"I64":2},{"F64":0.5}]`, the #155 wire form `Scalar` derives via serde). The cells
+/// bind positionally against the loaded signal's `param_space`. A malformed array is
+/// refused with the flag named, never silently coerced to an empty param vector
+/// (refuse-don't-guess, C10) — a dropped param would bootstrap a *different* graph.
+fn parse_param_cells(json: &str) -> Result<Vec<Scalar>, String> {
+    serde_json::from_str(json).map_err(|e| format!("--params: {e}"))
+}
+
+/// Parse the blueprint-run tail: `[--params <json>] [--seed <n>] [--real <SYMBOL>
+/// [--from <ms>] [--to <ms>]]` in any order, each flag at most once. `--from`/`--to` are
+/// legal only with `--real`; absent `--real` the run drives the built-in synthetic
+/// stream. Pure (no I/O, no exit) so the grammar is unit-testable; `main` does the side
+/// effects.
+fn parse_blueprint_run_args(rest: &[&str]) -> Result<BlueprintRunArgs, String> {
+    let usage = || {
+        "usage: aura run <blueprint.json> [--params <json-cell-array>] [--seed <n>] \
+         [--real <SYMBOL> [--from <ms>] [--to <ms>]]"
+            .to_string()
+    };
+    let mut params: Option<Vec<Scalar>> = None;
+    let mut seed: Option<u64> = None;
+    let mut symbol: Option<String> = None;
+    let mut from: Option<i64> = None;
+    let mut to: Option<i64> = None;
+    let mut tail = rest;
+    while let Some((flag, t)) = tail.split_first() {
+        match *flag {
+            "--params" if params.is_none() => {
+                let (value, t) = t.split_first().ok_or_else(usage)?;
+                params = Some(parse_param_cells(value)?);
+                tail = t;
+            }
+            "--seed" if seed.is_none() => {
+                let (value, t) = t.split_first().ok_or_else(usage)?;
+                seed = Some(value.parse().map_err(|_| usage())?);
+                tail = t;
+            }
+            "--real" if symbol.is_none() => {
+                let (value, t) = t.split_first().ok_or_else(usage)?;
+                if value.is_empty() {
+                    return Err(usage());
+                }
+                symbol = Some((*value).to_string());
+                tail = t;
+            }
+            "--from" if from.is_none() => {
+                let (value, t) = t.split_first().ok_or_else(usage)?;
+                from = Some(value.parse().map_err(|_| usage())?);
+                tail = t;
+            }
+            "--to" if to.is_none() => {
+                let (value, t) = t.split_first().ok_or_else(usage)?;
+                to = Some(value.parse().map_err(|_| usage())?);
+                tail = t;
+            }
+            _ => return Err(usage()),
+        }
+    }
+    // --from/--to require --real (mirrors the harness-kind run path's data-window guard).
+    if symbol.is_none() && (from.is_some() || to.is_some()) {
+        return Err(usage());
+    }
+    let data = match symbol {
+        Some(s) => RunData::Real { symbol: s, from, to },
+        None => RunData::Synthetic,
+    };
+    Ok(BlueprintRunArgs { params: params.unwrap_or_default(), data, seed: seed.unwrap_or(0) })
+}
+
 /// Route a parsed `run` invocation to its harness handler. The compile-time `match` IS
 /// the selector — the harnesses are Rust-authored built-ins, picked by name (C9/C17).
 fn run_dispatch(args: RunArgs) -> Result<RunReport, String> {
@@ -3291,19 +3566,47 @@ fn main() {
         return;
     }
     match args.iter().map(String::as_str).collect::<Vec<_>>().as_slice() {
-        ["run", rest @ ..] => match parse_run_args(rest) {
-            Ok(args) => match run_dispatch(args) {
-                Ok(report) => println!("{}", report.to_json()),
+        ["run", rest @ ..] => {
+            // `aura run <blueprint.json> [--params <json>] [--seed <n>] [--real ...]`: if the
+            // first positional is an existing `.json` file, load it as a serialized C24 signal
+            // blueprint (a `price`→`bias` leg; sinks/brokers/data are out of the round-trippable
+            // set), wrap it in the shared Stage-1-R scaffolding, and run. Otherwise fall through
+            // to the built-in harness-kind dispatch (#159 stays). The `.json`-file discriminator
+            // keeps the two paths disjoint — a bare `--harness` token never ends in `.json`.
+            if let Some(path) =
+                rest.first().filter(|a| a.ends_with(".json") && std::path::Path::new(a).is_file())
+            {
+                let doc = std::fs::read_to_string(path).unwrap_or_else(|e| {
+                    eprintln!("aura: {path}: {e}");
+                    std::process::exit(2);
+                });
+                let signal = blueprint_from_json(&doc, &|t| std_vocabulary(t)).unwrap_or_else(|e| {
+                    eprintln!("aura: {path}: {e:?}");
+                    std::process::exit(2);
+                });
+                let BlueprintRunArgs { params, data, seed } = parse_blueprint_run_args(&rest[1..])
+                    .unwrap_or_else(|msg| {
+                        eprintln!("aura: {msg}");
+                        std::process::exit(2);
+                    });
+                let report = run_signal_stage1r(signal, &params, data, seed);
+                println!("{}", report.to_json());
+                return;
+            }
+            match parse_run_args(rest) {
+                Ok(args) => match run_dispatch(args) {
+                    Ok(report) => println!("{}", report.to_json()),
+                    Err(msg) => {
+                        eprintln!("aura: {msg}");
+                        std::process::exit(2);
+                    }
+                },
                 Err(msg) => {
                     eprintln!("aura: {msg}");
                     std::process::exit(2);
                 }
-            },
-            Err(msg) => {
-                eprintln!("aura: {msg}");
-                std::process::exit(2);
             }
-        },
+        }
         ["chart", rest @ ..] => match parse_chart_args(rest) {
             Ok((name, tap, mode)) => emit_chart(&name, tap.as_deref(), mode),
             Err(msg) => {
@@ -4610,6 +4913,38 @@ mod tests {
         assert_eq!(a.cost, Some(0.001));
     }
 
+    #[test]
+    fn parse_param_cells_decodes_typed_cells_in_order_and_refuses_malformed() {
+        // The property: `--params` round-trips the externally-tagged Scalar wire form in
+        // array order, and a malformed array is refused with the flag named — never
+        // silently dropped (a dropped cell would compile-bind a different graph).
+        let cells = parse_param_cells("[{\"I64\":2},{\"F64\":0.5}]").expect("valid cell array");
+        assert_eq!(cells, vec![Scalar::I64(2), Scalar::F64(0.5)]);
+        assert!(parse_param_cells("[]").expect("empty array").is_empty());
+        let err = parse_param_cells("{not an array}").unwrap_err();
+        assert!(err.contains("--params"), "a malformed --params value names the flag: {err}");
+    }
+
+    #[test]
+    fn parse_blueprint_run_args_defaults_composes_and_guards() {
+        // The property: the blueprint-run tail defaults to empty params / synthetic data /
+        // seed 0, composes the flags in any order, and guards --from/--to behind --real.
+        let d = parse_blueprint_run_args(&[]).expect("bare blueprint run");
+        assert!(d.params.is_empty());
+        assert!(matches!(d.data, RunData::Synthetic));
+        assert_eq!(d.seed, 0);
+        let a = parse_blueprint_run_args(&[
+            "--seed", "7", "--params", "[{\"I64\":3}]", "--real", "GER40", "--from", "100",
+        ])
+        .expect("composed in any order");
+        assert_eq!(a.seed, 7);
+        assert_eq!(a.params, vec![Scalar::I64(3)]);
+        assert!(matches!(a.data, RunData::Real { ref symbol, from: Some(100), to: None } if symbol == "GER40"));
+        assert!(parse_blueprint_run_args(&["--from", "1"]).is_err()); // --from without --real
+        assert!(parse_blueprint_run_args(&["--seed", "x"]).is_err()); // non-numeric seed
+        assert!(parse_blueprint_run_args(&["--params", "[{\"I64\":1}]", "--params", "[]"]).is_err()); // repeated
+    }
+
     #[test]
     fn parse_mc_args_bare_and_name_route_to_synthetic() {
         assert_eq!(parse_mc_args(&[]), Ok(McArgs::Synthetic { name: "mc".to_string(), persist: false }));

crates/aura-engine/src/blueprint.rs

@@ -1002,6 +1002,7 @@ mod tests {
                 broker: "test".to_string(),
                 selection: None,
                 instrument: None,
+                topology_hash: None,
             },
             metrics: summarize(&equity, &exposure),
         }

crates/aura-engine/src/mc.rs

@@ -244,6 +244,7 @@ mod tests {
                 broker: "test".to_string(),
                 selection: None,
                 instrument: None,
+                topology_hash: None,
             },
             metrics: summarize(&equity, &exposure),
         }
@@ -269,6 +270,7 @@ mod tests {
                         broker: "t".to_string(),
                         selection: None,
                         instrument: None,
+                        topology_hash: None,
                     },
                     metrics: RunMetrics {
                         total_pips: v,

crates/aura-engine/src/report.rs

@@ -51,6 +51,12 @@ pub struct RunManifest {
     /// one-directional serde widening as `selection` (C14/C18).
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub instrument: Option<String>,
+    /// SHA256 (hex) of the canonical `blueprint_to_json` of the run's signal
+    /// topology — the #158 reproducibility anchor. `None` for a run not built
+    /// from a serialized blueprint and for every pre-0092 line. One-directional
+    /// serde widening (Tier-1, #156), identical idiom to `selection` / `instrument`.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub topology_hash: Option<String>,
 }
 
 /// A run's full structured result: the descriptor plus the metrics it
@@ -273,6 +279,7 @@ mod tests {
         let m = RunManifest {
             commit: "c".into(), params: vec![], window: (Timestamp(0), Timestamp(0)),
             seed: 0, broker: "b".into(), selection: None, instrument: None,
+            topology_hash: None,
         };
         assert!(!serde_json::to_string(&m).unwrap().contains("selection"));
     }
@@ -290,6 +297,7 @@ mod tests {
         let with = RunManifest {
             commit: "c".into(), params: vec![], window: (Timestamp(0), Timestamp(0)),
             seed: 0, broker: "b".into(), selection: None, instrument: Some("GER40".into()),
+            topology_hash: None,
         };
         let json = serde_json::to_string(&with).unwrap();
         assert!(json.contains("\"instrument\":\"GER40\""), "json: {json}");
@@ -299,6 +307,7 @@ mod tests {
         let without = RunManifest {
             commit: "c".into(), params: vec![], window: (Timestamp(0), Timestamp(0)),
             seed: 0, broker: "b".into(), selection: None, instrument: None,
+            topology_hash: None,
         };
         // skip_serializing_if omits the key when None -> existing manifest bytes unchanged (C23).
         assert!(!serde_json::to_string(&without).unwrap().contains("instrument"));
@@ -316,6 +325,7 @@ mod tests {
         let m = RunManifest {
             commit: "c".into(), params: vec![], window: (Timestamp(0), Timestamp(0)),
             seed: 0, broker: "b".into(), selection: Some(sel.clone()), instrument: None,
+            topology_hash: None,
         };
         let back: RunManifest = serde_json::from_str(&serde_json::to_string(&m).unwrap()).unwrap();
         assert_eq!(back.selection, Some(sel));
@@ -455,6 +465,7 @@ mod tests {
                 broker: "sim-optimal(pip_size=0.0001)".to_string(),
                 selection: None,
                 instrument: None,
+                topology_hash: None,
             },
             metrics,
         }
@@ -559,6 +570,7 @@ mod tests {
                 broker: "sim-optimal(pip_size=1.0)".to_string(),
                 selection: None,
                 instrument: None,
+                topology_hash: None,
             },
             metrics: RunMetrics {
                 total_pips: 12.0,
@@ -589,6 +601,7 @@ mod tests {
                 broker: "sim-optimal(pip_size=1.0)".to_string(),
                 selection: None,
                 instrument: None,
+                topology_hash: None,
             },
             metrics: RunMetrics { total_pips: 12.0, max_drawdown: 1.0, bias_sign_flips: 1, r: None },
         };
@@ -611,6 +624,7 @@ mod tests {
                 broker: "sim-optimal(pip_size=1.0)".to_string(),
                 selection: None,
                 instrument: None,
+                topology_hash: None,
             },
             metrics: RunMetrics { total_pips: 12.0, max_drawdown: 1.0, bias_sign_flips: 1, r: None },
         };

crates/aura-engine/src/sweep.rs

@@ -669,6 +669,7 @@ mod tests {
                 broker: "test".to_string(),
                 selection: None,
                 instrument: None,
+                topology_hash: None,
             },
             metrics: summarize(&equity, &exposure),
         }

crates/aura-engine/src/walkforward.rs

@@ -293,6 +293,7 @@ mod tests {
                 broker: "t".to_string(),
                 selection: None,
                 instrument: None,
+                topology_hash: None,
             },
             metrics: summarize(&[], &[]),
         }

crates/aura-engine/tests/random_sweep_e2e.rs

@@ -117,6 +117,7 @@ fn run_point(point: &[Cell]) -> RunReport {
             broker: "test".to_string(),
             selection: None,
             instrument: None,
+            topology_hash: None,
         },
         metrics: summarize(&equity, &exposure),
     }

crates/aura-ingest/examples/ger40_breakout_sweep.rs

@@ -68,6 +68,7 @@ fn run_point(server: &Arc<DataServer>, point: &[Cell], from: Timestamp, to: Time
             broker: "sim-optimal(pip_size=1)".to_string(),
             selection: None,
             instrument: None,
+            topology_hash: None,
         },
         metrics: summarize(&equity, &exposure),
     }

crates/aura-ingest/examples/ger40_breakout_walkforward.rs

@@ -78,6 +78,7 @@ fn run_point(
             broker: "sim-optimal(pip_size=1)".to_string(),
             selection: None,
             instrument: None,
+            topology_hash: None,
         },
         metrics: summarize(&equity, &exposure),
     };

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

@@ -401,6 +401,7 @@ pub fn report_from_trace(trace: &[BarTrace], from: Timestamp, to: Timestamp) ->
             broker: "sim-optimal(pip_size=1)".to_string(),
             selection: None,
             instrument: None,
+            topology_hash: None,
         },
         metrics,
     }

crates/aura-ingest/tests/ger40_breakout_world.rs

@@ -70,6 +70,7 @@ fn run_point(server: &Arc<DataServer>, point: &[Cell], from: Timestamp, to: Time
             broker: "sim-optimal(pip_size=1)".to_string(),
             selection: None,
             instrument: None,
+            topology_hash: None,
         },
         metrics: summarize(&equity, &exposure),
     }

crates/aura-ingest/tests/real_bars.rs

@@ -91,6 +91,7 @@ fn run_sample_over(prices: Vec<(Timestamp, Scalar)>) -> RunReport {
             broker: "sim-optimal(pip_size=0.0001)".to_string(),
             selection: None,
             instrument: None,
+            topology_hash: None,
         },
         metrics,
     }

crates/aura-ingest/tests/streaming_seam.rs

@@ -91,6 +91,7 @@ fn run_sample(window: (Timestamp, Timestamp), source: Box<dyn Source>) -> RunRep
             broker: "sim-optimal(pip_size=0.0001)".to_string(),
             selection: None,
             instrument: None,
+            topology_hash: None,
         },
         metrics: summarize(&equity, &exposure),
     }

crates/aura-registry/src/compat.rs

@@ -34,6 +34,8 @@ struct RunManifestRead {
     selection: Option<FamilySelection>,
     #[serde(default)]
     instrument: Option<String>,
+    #[serde(default)]
+    topology_hash: Option<String>,
 }
 
 /// A param value that accepts BOTH the current tagged [`Scalar`] form
@@ -66,7 +68,8 @@ impl<'de> Deserialize<'de> for ScalarRead {
 
 impl From<RunReportRead> for RunReport {
     fn from(r: RunReportRead) -> Self {
-        let RunManifestRead { commit, params, window, seed, broker, selection, instrument } = r.manifest;
+        let RunManifestRead { commit, params, window, seed, broker, selection, instrument, topology_hash } =
+            r.manifest;
         RunReport {
             manifest: RunManifest {
                 commit,
@@ -76,6 +79,7 @@ impl From<RunReportRead> for RunReport {
                 broker,
                 selection,
                 instrument,
+                topology_hash,
             },
             metrics: r.metrics,
         }

crates/aura-registry/src/lib.rs

@@ -551,6 +551,7 @@ mod tests {
                 broker: "b".to_string(),
                 selection: None,
                 instrument: None,
+                topology_hash: None,
             },
             metrics: RunMetrics { total_pips, max_drawdown, bias_sign_flips: flips, r: None },
         }
@@ -886,6 +887,7 @@ mod tests {
                 manifest: RunManifest {
                     commit: "c".into(), params: vec![], window: (Timestamp(0), Timestamp(0)),
                     seed: 0, broker: "b".into(), selection: None, instrument: None,
+                    topology_hash: None,
                 },
                 metrics: RunMetrics {
                     total_pips, max_drawdown: 0.0, bias_sign_flips: 0,

crates/aura-registry/src/trace_store.rs

@@ -255,6 +255,7 @@ mod tests {
             broker: "sim-optimal(pip_size=0.0001)".to_string(),
             selection: None,
             instrument: None,
+            topology_hash: None,
         }
     }