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
2026-06-30 20:23:42 +02:00
parent e25ca15be4
commit ff4a1b3d4a
18 changed files with 454 additions and 69 deletions
@@ -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 }));