fix(aura-runner, aura-cli): the reference hash reaches the run before persistence

Harvest sweep, review re-check fix.

The prior fix corrected only the stdout record line: run_signal_r
persists the manifest to runs/traces/<name>/index.json (bound.finish)
BEFORE the post-hoc overwrite ran, so an overridden run's trace store
still carried the reopened hash — diverging from stdout in exactly the
case the revised C24/C18 clause describes. run_signal_r now takes
topo: Option<&str> (None = inline computation as before; Some = the
caller's reference-semantics hash, the run_blueprint_member precedent)
so record line and trace index read the one hash built before any
persistence; exec's override branch passes the base document's id and
the post-run mutation is gone. RED-first: the new trace-store pin
failed against the divergent state.

refs #343

crates/aura-cli/src/main.rs

@@ -1298,7 +1298,12 @@ fn exec_blueprint_leg(path: &str, overrides: &[String], tap: &[String], env: &au
         // precedent (`runner.rs::MemberRunner::run_member` passes
         // `&cell.strategy_id`, the stored base document's id, as `topo` even
         // for a reopened member). Computed here, over the still-unreopened
-        // `signal`, before `reopen_all` consumes it.
+        // `signal`, before `reopen_all` consumes it, and threaded into
+        // `run_signal_r`'s own `topo` parameter so every consumer inside that
+        // function — the record line AND the trace-store persistence
+        // (`bound.finish`, `aura-registry::trace_store`) — sees the base
+        // document's hash from the start; no post-hoc manifest mutation
+        // after the run, which would miss the already-persisted `index.json`.
         let base_topo =
             content_id(&blueprint_to_json(&signal).expect("a buildable signal serializes"));
         let signal = reopen_all(signal, &paths);
@@ -1312,18 +1317,13 @@ fn exec_blueprint_leg(path: &str, overrides: &[String], tap: &[String], env: &au
         // `SilencedPanic` + `catch_unwind`) and render it as a runtime-class
         // refusal (exit 1, C14 partition) instead of an uncaught exit 101 —
         // the input was well-formed argv; the value is what the node refuses.
-        let mut report = aura_campaign::catch_member_panic(|| {
-            run_signal_r(signal, &params, RunData::Synthetic, 0, env, tap_plan)
+        let report = aura_campaign::catch_member_panic(|| {
+            run_signal_r(signal, &params, RunData::Synthetic, 0, env, tap_plan, Some(&base_topo))
         })
         .unwrap_or_else(|msg| {
             eprintln!("aura: {}", panic_refusal_prose(&msg));
             std::process::exit(1);
         });
-        // Overwrite `run_signal_r`'s own (reopened-topology) hash with the
-        // base document's id: reference semantics (#343). The manifest's
-        // `params` already carries the override as the variation; the base
-        // document + params deterministically reconstruct this run.
-        report.manifest.topology_hash = Some(base_topo);
         // #324 (comment 4501): this leg always runs `RunData::Synthetic` (no
         // direct-blueprint exec ever binds real data), so a validating
         // caller reading zero trades here as "broken strategy" needs telling

crates/aura-cli/tests/exec.rs

@@ -612,6 +612,52 @@ base {base_out} / overridden {ov_out}"
     );
 }
 
+/// Follow-up to the #343 revised pin above: `run_signal_r`'s in-graph tap
+/// persistence (`bound.finish(&manifest)`, `aura-registry::trace_store`)
+/// writes `traces/<name>/index.json` INSIDE `run_signal_r`, before
+/// `exec_blueprint_leg` ever sees the returned `RunReport` — so the
+/// persisted manifest must ALSO carry the base document's reference-semantics
+/// hash, not just the stdout record line. A tap-bearing blueprint (so
+/// `index.json` exists to inspect) with a no-op `--override` pins that the
+/// override run's persisted `topology_hash` equals both its own stdout hash
+/// and the un-overridden base run's hash.
+#[test]
+fn exec_blueprint_override_persists_the_base_documents_hash_in_the_trace_store() {
+    let cwd = temp_cwd("override-persisted-hash");
+    let bp_path = cwd.join("tapped_r_sma.json");
+    std::fs::write(&bp_path, tap_blueprint_json()).expect("write tapped blueprint");
+
+    let (base_out, base_code) =
+        run_code_in(&cwd, &["exec", bp_path.to_str().expect("utf-8 path")]);
+    assert_eq!(base_code, Some(0), "stdout/stderr: {base_out}");
+    let base_line = base_out.lines().find(|l| l.starts_with('{')).expect("record line");
+    let base_v: serde_json::Value = serde_json::from_str(base_line).expect("record parses");
+    let base_hash =
+        base_v["manifest"]["topology_hash"].as_str().expect("base topology_hash").to_string();
+
+    let (ov_out, ov_code) = run_code_in(
+        &cwd,
+        &["exec", bp_path.to_str().expect("utf-8 path"), "--override", "fast.length=2"],
+    );
+    assert_eq!(ov_code, Some(0), "stdout/stderr: {ov_out}");
+    let ov_line = ov_out.lines().find(|l| l.starts_with('{')).expect("record line");
+    let ov_v: serde_json::Value = serde_json::from_str(ov_line).expect("record parses");
+    let ov_stdout_hash =
+        ov_v["manifest"]["topology_hash"].as_str().expect("overridden topology_hash").to_string();
+    assert_eq!(ov_stdout_hash, base_hash, "stdout already carries the base document's hash");
+
+    let index_text = std::fs::read_to_string(cwd.join("runs/traces/sma_signal/index.json"))
+        .expect("read index.json");
+    let index: serde_json::Value = serde_json::from_str(&index_text).expect("parse index.json");
+    let persisted_hash =
+        index["manifest"]["topology_hash"].as_str().expect("persisted topology_hash").to_string();
+    assert_eq!(
+        persisted_hash, base_hash,
+        "the trace store's persisted manifest must carry the SAME reference-semantics hash \
+as stdout, not the reopened topology's own — index.json: {index_text}"
+    );
+}
+
 /// A malformed override token (no `NODE.PARAM=VALUE` shape) refuses as usage.
 #[test]
 fn exec_override_malformed_token_refuses_with_usage() {

crates/aura-runner/src/member.rs

@@ -519,17 +519,33 @@ fn compile_error_prose(e: &CompileError, names: &[String], params: &[Scalar]) ->
 /// 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.
+///
+/// `topo`: `None` computes `topology_hash` inline from `signal` as always
+/// (every existing caller); `Some(hash)` overrides it with the caller's own
+/// **reference-semantics** hash (#343, revised) — the exec blueprint leg's
+/// `--override` branch passes the loaded base document's content id, computed
+/// BEFORE `reopen_all`, so the manifest reaching every consumer inside this
+/// function (the record line AND the trace-store persistence below) carries
+/// the base document's id, never the reopened topology's own, exactly as the
+/// campaign leg's `run_blueprint_member` has always taken its `topo` as a
+/// caller-supplied reference (`runner.rs`'s `&cell.strategy_id`).
 #[allow(clippy::type_complexity)]
 pub fn run_signal_r(
     signal: Composite, params: &[Scalar], data: RunData, seed: u64, env: &Env,
-    plan: TapPlan,
+    plan: TapPlan, topo: Option<&str>,
 ) -> RunReport {
     // topology_hash's own two-line body, inlined: `content_id_of` over the
     // canonical (#164) blueprint JSON — the CLI shell's `topology_hash`
     // helper is the same primitive, kept single-sourced at `aura_research`.
-    let topo = aura_research::content_id_of(
-        &aura_engine::blueprint_to_json(&signal).expect("a buildable signal serializes"),
-    ); // before signal is consumed
+    // Skipped when the caller already supplies a reference-semantics hash
+    // (`topo: Some(..)`) — no need to hash a signal whose own topology_hash
+    // will be overridden anyway.
+    let topo: String = match topo {
+        Some(t) => t.to_string(),
+        None => aura_research::content_id_of(
+            &aura_engine::blueprint_to_json(&signal).expect("a buildable signal serializes"),
+        ), // before signal is consumed
+    };
     let run_name = signal.name().to_string(); // before signal is consumed by `wrap_r`
     // The default binding (name defaults; `aura run` carries no campaign
     // overrides). Refusals are the established `aura: ` + exit-1 register.
@@ -1208,8 +1224,8 @@ mod tests {
         let env = Env::std();
         let loaded = blueprint_from_json(include_str!("../../aura-cli/examples/r_breakout.json"), &|t| std_vocabulary(t))
             .expect("shipped r_breakout example loads");
-        let via_file = run_signal_r(loaded, &[], RunData::Synthetic, 0, &env, TapPlan::record_all());
-        let via_carve = run_signal_r(r_breakout_signal(Some(R_BREAKOUT_CHANNEL)), &[], RunData::Synthetic, 0, &env, TapPlan::record_all());
+        let via_file = run_signal_r(loaded, &[], RunData::Synthetic, 0, &env, TapPlan::record_all(), None);
+        let via_carve = run_signal_r(r_breakout_signal(Some(R_BREAKOUT_CHANNEL)), &[], RunData::Synthetic, 0, &env, TapPlan::record_all(), None);
         assert_eq!(via_file.metrics, via_carve.metrics, "loaded example grades identically to the carve");
     }
 
@@ -1361,7 +1377,7 @@ mod tests {
         let env = Env::std();
         let loaded = blueprint_from_json(include_str!("../../aura-cli/examples/r_meanrev.json"), &|t| std_vocabulary(t))
             .expect("shipped r_meanrev example loads");
-        let via_file = run_signal_r(loaded, &[], RunData::Synthetic, 0, &env, TapPlan::record_all());
+        let via_file = run_signal_r(loaded, &[], RunData::Synthetic, 0, &env, TapPlan::record_all(), None);
         let via_carve = run_signal_r(
             r_meanrev_signal(Some(R_MEANREV_WINDOW), Some(R_MEANREV_BAND_K)),
             &[],
@@ -1369,6 +1385,7 @@ mod tests {
             0,
             &env,
             TapPlan::record_all(),
+            None,
         );
         assert_eq!(via_file.metrics, via_carve.metrics, "loaded example grades identically to the carve");
     }
@@ -1783,7 +1800,7 @@ mod tests {
     fn fold_and_live_plans_agree_with_the_recorded_series() {
         // Run A: record (the charting question).
         let (env_a, root_a) = temp_project_env("record");
-        run_signal_r(tapped_r_sma(), &[], RunData::Synthetic, 0, &env_a, TapPlan::record_all());
+        run_signal_r(tapped_r_sma(), &[], RunData::Synthetic, 0, &env_a, TapPlan::record_all(), None);
         let series = read_tap_json(&root_a);
         let ts: Vec<i64> =
             series["ts"].as_array().unwrap().iter().map(|v| v.as_i64().unwrap()).collect();
@@ -1795,7 +1812,7 @@ mod tests {
         let (env_b, root_b) = temp_project_env("fold");
         let mut plan_b = TapPlan::record_all();
         plan_b.subscribe("fast_tap", TapSubscription::named("mean"));
-        run_signal_r(tapped_r_sma(), &[], RunData::Synthetic, 0, &env_b, plan_b);
+        run_signal_r(tapped_r_sma(), &[], RunData::Synthetic, 0, &env_b, plan_b, None);
         let row = read_tap_json(&root_b);
         let mean = vals.iter().sum::<f64>() / vals.len() as f64; // sequential — FoldState's order
         assert_eq!(row["ts"], serde_json::json!([*ts.last().unwrap()]));
@@ -1811,7 +1828,7 @@ mod tests {
                 let _ = live_tx.send((ts.0, cell.f64()));
             }),
         );
-        run_signal_r(tapped_r_sma(), &[], RunData::Synthetic, 0, &env_c, plan_c);
+        run_signal_r(tapped_r_sma(), &[], RunData::Synthetic, 0, &env_c, plan_c, None);
         let got: Vec<(i64, f64)> = live_rx.try_iter().collect();
         let want: Vec<(i64, f64)> = ts.iter().copied().zip(vals.iter().copied()).collect();
         assert_eq!(got, want, "the live closure saw exactly the recorded series (C1)");
@@ -1862,8 +1879,8 @@ mod tests {
     fn two_identical_record_runs_produce_byte_identical_tap_files() {
         let (env_a, root_a) = temp_project_env("det-a");
         let (env_b, root_b) = temp_project_env("det-b");
-        run_signal_r(tapped_r_sma(), &[], RunData::Synthetic, 0, &env_a, TapPlan::record_all());
-        run_signal_r(tapped_r_sma(), &[], RunData::Synthetic, 0, &env_b, TapPlan::record_all());
+        run_signal_r(tapped_r_sma(), &[], RunData::Synthetic, 0, &env_a, TapPlan::record_all(), None);
+        run_signal_r(tapped_r_sma(), &[], RunData::Synthetic, 0, &env_b, TapPlan::record_all(), None);
         let a = std::fs::read(
             root_a.join("runs").join("traces").join("sma_signal").join("fast_tap.json"),
         )