feat(aura-runner, aura-cli): run_measurement on the shared tap pair; integration + register coverage (closes #283, closes #77)

Completes the #283 registry cycle. run_measurement drops its
duplicated Recorder-bind + post-run try_iter drain for the shared
bind_tap_plan/BoundTaps pair — the mirror of run_signal_r is now the
same code, so the two declared-tap entry points cannot drift; the
measure CLI arm passes TapPlan::record_all() (semantics unchanged,
constant memory). Measure-side library tests pin a fold plan's
one-row trace (streamed mean bit-equal to the slice mean over the
synthetic prices) and the tap-free write-nothing guarantee through
the shared pair.

Run-side integration tests pin the #283 acceptance surface: fold and
live plans agree exactly with the recorded series of the same
blueprint (C1, three runs, three drain policies); a live-only plan
persists nothing; plan refusals (unknown tap, roster-enumerating
unknown label) fire typed before any store I/O; two identical
record-all runs produce byte-identical tap files. The previously
untested `aura: writing tap traces failed` register gains end-to-end
coverage on both arms: store root occupied by a file (pre-run
refusal) and a read-only run dir (deferred initialize-open failure —
the run completes, the failure surfaces terminally, no index.json is
written, the store's treat-as-absent crash shape).

Two forced deviations from the plan's scripted literals, both
verified equivalent: an explicit match replaces expect_err (BoundTaps
is deliberately not Debug — it carries a TraceStreamer), and the test
helper takes &Path, not &PathBuf (clippy::ptr_arg under -D warnings).

Closes #77 as specced: zero per-cycle heap on the tap path
((Timestamp, Cell) payload, committed earlier this branch) and the
Recorder→Probe rename retired (no Probe symbol anywhere; decision
recorded on #77, 2026-07-21).

Verification: cargo test --workspace — 1517 passed, 0 failed;
cargo clippy --workspace --all-targets -- -D warnings clean;
tap_recording 5/5 (byte-compat anchors + both register arms).

refs #283 refs #77

crates/aura-cli/src/main.rs

@@ -1787,7 +1787,7 @@ fn dispatch_run(a: RunCmd, env: &aura_runner::project::Env) {
                     None => Vec::new(),
                 };
                 let data = run_data_from(a.real.as_deref(), a.from, a.to);
-                let report = run_measurement(signal, &params, data, a.seed.unwrap_or(0), env);
+                let report = run_measurement(signal, &params, data, a.seed.unwrap_or(0), env, TapPlan::record_all());
                 println!("{}", report.to_json());
             } else {
                 eprintln!(

crates/aura-cli/tests/tap_recording.rs

@@ -150,3 +150,65 @@ fn single_run_refuses_a_duplicate_tap_name_before_persisting_anything() {
     );
     assert!(!cwd.join("runs").exists(), "a refused run must not persist a half-written trace store");
 }
+
+/// The `aura: writing tap traces failed: …` register, pre-run arm: `runs`
+/// exists as a FILE, so the trace store's directory creation
+/// (`begin_run`) fails before the run — exit 1 through the register.
+/// (This register was code-present but untested before #283.)
+#[test]
+fn an_unwritable_store_root_exits_through_the_tap_trace_register() {
+    let cwd = temp_cwd("store-root-is-a-file");
+    let bp_path = cwd.join("tapped_r_sma.json");
+    std::fs::write(&bp_path, tap_blueprint_json()).expect("write tapped blueprint");
+    std::fs::write(cwd.join("runs"), b"not a directory").expect("occupy runs as a file");
+
+    let out = Command::new(BIN)
+        .args(["run", bp_path.to_str().expect("utf-8 path")])
+        .current_dir(&cwd)
+        .output()
+        .expect("spawn aura run");
+
+    assert!(!out.status.success(), "an unwritable store must refuse, not succeed");
+    let stderr = String::from_utf8_lossy(&out.stderr);
+    assert!(
+        stderr.contains("writing tap traces failed"),
+        "the tap-trace register must fire: {stderr}"
+    );
+}
+
+/// The same register, deferred arm (#283): the run directory pre-exists
+/// read-only, so `begin_run` succeeds (create_dir_all on an existing dir)
+/// but the record consumer's deferred open in `initialize` fails; the run
+/// COMPLETES, the failure surfaces terminally through the register, and no
+/// `index.json` is written (the store's treat-as-absent crash shape).
+#[cfg(unix)]
+#[test]
+fn a_read_only_run_dir_surfaces_terminally_through_the_tap_trace_register() {
+    use std::os::unix::fs::PermissionsExt;
+
+    let cwd = temp_cwd("run-dir-read-only");
+    let bp_path = cwd.join("tapped_r_sma.json");
+    std::fs::write(&bp_path, tap_blueprint_json()).expect("write tapped blueprint");
+    let run_dir = cwd.join("runs/traces/sma_signal");
+    std::fs::create_dir_all(&run_dir).expect("pre-create run dir");
+    std::fs::set_permissions(&run_dir, std::fs::Permissions::from_mode(0o555))
+        .expect("make run dir read-only");
+
+    let out = Command::new(BIN)
+        .args(["run", bp_path.to_str().expect("utf-8 path")])
+        .current_dir(&cwd)
+        .output()
+        .expect("spawn aura run");
+
+    // restore permissions FIRST so the next run's temp_cwd cleanup works.
+    std::fs::set_permissions(&run_dir, std::fs::Permissions::from_mode(0o755))
+        .expect("restore run dir permissions");
+
+    assert!(!out.status.success(), "a failed writer must surface, not pass silently");
+    let stderr = String::from_utf8_lossy(&out.stderr);
+    assert!(
+        stderr.contains("writing tap traces failed"),
+        "the tap-trace register must fire terminally: {stderr}"
+    );
+    assert!(!run_dir.join("index.json").exists(), "no index — treat-as-absent crash shape");
+}

crates/aura-runner/src/measure.rs

@@ -7,15 +7,14 @@
 //! into an `IcReport`) is unrelated to this run path and stays in the CLI
 //! shell (`dispatch_measure_ic`), which assembles and prints that DTO.
 
-use std::collections::{BTreeMap, BTreeSet};
-use std::sync::mpsc;
+use std::collections::BTreeMap;
 
-use aura_core::{Firing, Scalar, ScalarKind, Timestamp};
-use aura_engine::{ColumnarTrace, Composite, Harness, MeasurementReport, RunManifest, TapBindError};
-use aura_std::Recorder;
+use aura_core::{Scalar, Timestamp};
+use aura_engine::{Composite, Harness, MeasurementReport, RunManifest};
 
 use crate::member::{key_supply, resolve_run_data, wrapped_bound_defaults, RunData};
 use crate::project::Env;
+use crate::tap_plan::{bind_tap_plan, TapPlan};
 
 /// The single build-time commit provenance (`option_env!("AURA_COMMIT")`,
 /// falling back to `"unknown"`) — `measurement_manifest`'s `RunManifest.commit`
@@ -56,10 +55,11 @@ pub fn measurement_manifest(
 /// eq/ex/r R-evaluation, KEEPING the declared-tap bind → drain → persist (C27).
 /// No broker, no risk executor, no per-cycle equity/exposure/r recorders — this
 /// is where the measured O(cycles) retention is removed. The tap machinery is
-/// duplicated (not extracted) so `run_signal_r` stays byte-identical.
-#[allow(clippy::type_complexity)]
+/// the shared `bind_tap_plan`/`BoundTaps` pair — one wiring for both entry
+/// points, so they cannot drift (#283).
 pub fn run_measurement(
     signal: Composite, params: &[Scalar], data: RunData, seed: u64, env: &Env,
+    plan: TapPlan,
 ) -> MeasurementReport {
     // topology_hash's own two-line body, inlined (mirrors member::run_signal_r):
     // `content_id_of` over the canonical (#164) blueprint JSON — the CLI shell's
@@ -88,22 +88,12 @@ pub fn run_measurement(
         std::process::exit(1);
     });
 
-    // Bind each declared tap to a Recorder (mirrors run_signal_r).
-    let mut seen: BTreeSet<String> = BTreeSet::new();
-    let mut tap_drains: Vec<(String, ScalarKind, mpsc::Receiver<(Timestamp, Vec<Scalar>)>)> = Vec::new();
-    let declared: Vec<aura_engine::FlatTap> = flat.taps.clone();
-    for tap in &declared {
-        if !seen.insert(tap.name.clone()) {
-            eprintln!("aura: {}", TapBindError::DuplicateBind { name: tap.name.clone() });
-            std::process::exit(1);
-        }
-        let kind = flat.signatures[tap.node].output[tap.field].kind;
-        let (tx, rx) = mpsc::channel();
-        let sig = Recorder::builder(vec![kind], Firing::Any, tx.clone()).schema().clone();
-        flat.bind_tap(&tap.name, Box::new(Recorder::new(&[kind], Firing::Any, tx)), sig)
-            .expect("declared tap binds (name from flat.taps, kind from its own signature)");
-        tap_drains.push((tap.name.clone(), kind, rx));
-    }
+    // Bind each declared tap per the plan's subscription (mirrors
+    // run_signal_r — the shared bind_tap_plan/BoundTaps pair IS the mirror).
+    let bound = bind_tap_plan(&mut flat, plan, env, &run_name).unwrap_or_else(|e| {
+        eprintln!("aura: {e}");
+        std::process::exit(1);
+    });
 
     let mut h = Harness::bootstrap(flat).expect("valid measurement harness");
     h.run_bound(key_supply(&binding, sources))
@@ -116,20 +106,95 @@ pub fn run_measurement(
     manifest.topology_hash = Some(topo);
     manifest.project = env.provenance();
 
-    // Drain + persist each declared tap (mirrors run_signal_r).
-    let tap_names: Vec<String> = tap_drains.iter().map(|(n, _, _)| n.clone()).collect();
-    if !tap_drains.is_empty() {
-        let tap_traces: Vec<ColumnarTrace> = tap_drains
-            .into_iter()
-            .map(|(name, kind, rx)| {
-                let rows: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
-                ColumnarTrace::from_rows(&name, &[kind], &rows)
-            })
-            .collect();
-        env.trace_store().write(&run_name, &manifest, &tap_traces).unwrap_or_else(|e| {
-            eprintln!("aura: writing tap traces failed: {e}");
-            std::process::exit(1);
-        });
-    }
+    // Close the tap plan (mirrors run_signal_r; nothing buffered, #283).
+    let tap_names: Vec<String> = bound.declared_names().to_vec();
+    bound.finish(&manifest).unwrap_or_else(|e| {
+        eprintln!("aura: {e}");
+        std::process::exit(1);
+    });
     MeasurementReport { manifest, taps: tap_names }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::project::ProjectEnv;
+    use crate::tap_plan::TapSubscription;
+    use aura_engine::blueprint_from_json;
+    use aura_vocabulary::std_vocabulary;
+    use std::path::PathBuf;
+
+    /// A fresh project root so the run's `runs/` lands in a temp dir.
+    fn temp_project_env(name: &str) -> (Env, PathBuf) {
+        let root = std::env::temp_dir()
+            .join(format!("aura-measure-plan-{name}-{}", std::process::id()));
+        let _ = std::fs::remove_dir_all(&root);
+        std::fs::create_dir_all(&root).expect("create temp project root");
+        let env = Env::with_project(ProjectEnv {
+            root: root.clone(),
+            toml: Default::default(),
+            commit: None,
+            native: None,
+        });
+        (env, root)
+    }
+
+    /// The shipped `examples/r_sma.json` with one declared tap on node 0
+    /// ("fast", SMA(length=2)) field 0 — the same patch shape as
+    /// `aura-cli/tests/tap_recording.rs`.
+    fn tapped_r_sma() -> Composite {
+        let doc = include_str!("../../aura-cli/examples/r_sma.json");
+        let mut v: serde_json::Value = serde_json::from_str(doc).expect("parse r_sma.json");
+        v["blueprint"]["taps"] =
+            serde_json::json!([{"name": "fast_tap", "from": {"node": 0, "field": 0}}]);
+        let patched = serde_json::to_string(&v).expect("re-serialize");
+        blueprint_from_json(&patched, &|t| std_vocabulary(t)).expect("tapped r_sma loads")
+    }
+
+    /// The exact synthetic price array the `RunData::Synthetic` path feeds
+    /// (`r_sma_prices()`, main.rs) — duplicated so the fold value is pinned
+    /// against an independently-derived expectation, not a blind capture
+    /// (the `tap_recording.rs` discipline).
+    const R_SMA_PRICES: [f64; 18] = [
+        1.0000, 1.0008, 1.0021, 1.0039, 1.0062, 1.0090, 1.0083, 1.0061, 1.0034, 1.0012, 0.9998,
+        1.0006, 1.0024, 1.0047, 1.0069, 1.0086, 1.0097, 1.0092,
+    ];
+
+    #[test]
+    fn measurement_fold_plan_writes_one_row_trace_through_the_shared_pair() {
+        let (env, root) = temp_project_env("fold");
+        let mut plan = TapPlan::empty();
+        plan.subscribe("fast_tap", TapSubscription::named("mean"));
+        let report = run_measurement(tapped_r_sma(), &[], RunData::Synthetic, 0, &env, plan);
+        assert_eq!(report.taps, vec!["fast_tap".to_string()]);
+
+        let text = std::fs::read_to_string(
+            root.join("runs").join("traces").join("sma_signal").join("fast_tap.json"),
+        )
+        .expect("one-row fold trace persisted");
+        let v: serde_json::Value = serde_json::from_str(&text).expect("parse");
+        assert_eq!(v["kinds"], serde_json::json!(["F64"]));
+        let ts = v["ts"].as_array().expect("ts");
+        let col = v["columns"][0].as_array().expect("col");
+        assert_eq!((ts.len(), col.len()), (1, 1), "exactly one summary row");
+        // SMA(2) series (warm from sample 2), folded mean — sequential sum,
+        // same operation order as FoldState (bit-identical, IEEE-754).
+        let sma: Vec<f64> = (1..R_SMA_PRICES.len())
+            .map(|i| (R_SMA_PRICES[i - 1] + R_SMA_PRICES[i]) / 2.0)
+            .collect();
+        let mean = sma.iter().sum::<f64>() / sma.len() as f64;
+        assert_eq!(col[0].as_f64().expect("f64 row"), mean, "streamed mean == slice mean");
+        assert_eq!(ts[0].as_i64().expect("ts"), R_SMA_PRICES.len() as i64, "last folded ts");
+    }
+
+    #[test]
+    fn measurement_tap_free_blueprint_writes_nothing() {
+        let (env, root) = temp_project_env("tapfree");
+        let doc = include_str!("../../aura-cli/examples/r_sma.json");
+        let signal = blueprint_from_json(doc, &|t| std_vocabulary(t)).expect("r_sma loads");
+        let report =
+            run_measurement(signal, &[], RunData::Synthetic, 0, &env, TapPlan::record_all());
+        assert!(report.taps.is_empty(), "no declared taps");
+        assert!(!root.join("runs").exists(), "a tap-free run writes no runs/ entry");
+    }
+}

crates/aura-runner/src/member.rs

@@ -935,6 +935,9 @@ mod tests {
     use aura_engine::blueprint_to_json;
     use aura_strategy::{ConstantCost, VolSlippageCost};
     use aura_vocabulary::std_vocabulary;
+    use crate::tap_plan::{bind_tap_plan, TapPlanError, TapSubscription};
+    use crate::project::ProjectEnv;
+    use std::path::{Path, PathBuf};
 
     #[test]
     /// Independently pins the shipped `r_meanrev_signal` carve's FADE direction —
@@ -1628,4 +1631,133 @@ mod tests {
         );
     }
 
+    /// A fresh project root so a run's `runs/` lands in a temp dir.
+    fn temp_project_env(name: &str) -> (Env, PathBuf) {
+        let root = std::env::temp_dir()
+            .join(format!("aura-member-plan-{name}-{}", std::process::id()));
+        let _ = std::fs::remove_dir_all(&root);
+        std::fs::create_dir_all(&root).expect("create temp project root");
+        let env = Env::with_project(ProjectEnv {
+            root: root.clone(),
+            toml: Default::default(),
+            commit: None,
+            native: None,
+        });
+        (env, root)
+    }
+
+    /// The shipped `examples/r_sma.json` with one declared tap on node 0
+    /// ("fast", SMA(length=2)) field 0 — the `tap_recording.rs` patch shape.
+    /// `Composite` is `!Clone`, so each run loads it afresh.
+    fn tapped_r_sma() -> Composite {
+        let doc = include_str!("../../aura-cli/examples/r_sma.json");
+        let mut v: serde_json::Value = serde_json::from_str(doc).expect("parse r_sma.json");
+        v["blueprint"]["taps"] =
+            serde_json::json!([{"name": "fast_tap", "from": {"node": 0, "field": 0}}]);
+        let patched = serde_json::to_string(&v).expect("re-serialize");
+        blueprint_from_json(&patched, &|t| std_vocabulary(t)).expect("tapped r_sma loads")
+    }
+
+    fn read_tap_json(root: &Path) -> serde_json::Value {
+        let text = std::fs::read_to_string(
+            root.join("runs").join("traces").join("sma_signal").join("fast_tap.json"),
+        )
+        .expect("persisted tap trace");
+        serde_json::from_str(&text).expect("parse tap trace")
+    }
+
+    /// #283 acceptance: a fold plan's one-row summary and a live plan's
+    /// collected stream both agree with the recorded series of the same
+    /// blueprint (C1 — three identical runs, three drain policies), and a
+    /// live-only plan persists nothing at all.
+    #[test]
+    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());
+        let series = read_tap_json(&root_a);
+        let ts: Vec<i64> =
+            series["ts"].as_array().unwrap().iter().map(|v| v.as_i64().unwrap()).collect();
+        let vals: Vec<f64> =
+            series["columns"][0].as_array().unwrap().iter().map(|v| v.as_f64().unwrap()).collect();
+        assert!(!vals.is_empty(), "the recorded series is the reference");
+
+        // Run B: fold to mean (the aggregate question, same blueprint).
+        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);
+        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()]));
+        assert_eq!(row["columns"], serde_json::json!([[mean]]), "fold row == slice mean, bit-exact");
+
+        // Run C: live only (nothing persisted; the closure sees the series).
+        let (env_c, root_c) = temp_project_env("live");
+        let (live_tx, live_rx) = std::sync::mpsc::channel();
+        let mut plan_c = TapPlan::empty();
+        plan_c.subscribe(
+            "fast_tap",
+            TapSubscription::live(move |ts, cell| {
+                let _ = live_tx.send((ts.0, cell.f64()));
+            }),
+        );
+        run_signal_r(tapped_r_sma(), &[], RunData::Synthetic, 0, &env_c, plan_c);
+        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)");
+        assert!(!root_c.join("runs").exists(), "a live-only plan persists nothing");
+    }
+
+    /// Plan refusals are typed and fire before any store I/O.
+    #[test]
+    fn bind_tap_plan_refuses_unknown_tap_and_unknown_label_before_the_store() {
+        let (env, root) = temp_project_env("refusals");
+
+        // Unknown tap name.
+        let mut flat = tapped_r_sma().compile_with_params(&[]).expect("tapped r_sma compiles");
+        let mut plan = TapPlan::record_all();
+        plan.subscribe("no_such_tap", TapSubscription::named("mean"));
+        // `BoundTaps` (the `Ok` side) carries no `Debug`, so `expect_err` is
+        // unavailable here — match explicitly instead (same refusal intent).
+        let err = match bind_tap_plan(&mut flat, plan, &env, "sma_signal") {
+            Err(e) => e,
+            Ok(_) => panic!("unknown tap must be refused"),
+        };
+        assert!(
+            matches!(err, TapPlanError::UnknownTap { ref name } if name == "no_such_tap"),
+            "{err:?}"
+        );
+
+        // Unknown label — the refusal enumerates the roster.
+        let mut flat2 = tapped_r_sma().compile_with_params(&[]).expect("tapped r_sma compiles");
+        let mut plan2 = TapPlan::record_all();
+        plan2.subscribe("fast_tap", TapSubscription::named("p95"));
+        let err2 = match bind_tap_plan(&mut flat2, plan2, &env, "sma_signal") {
+            Err(e) => e,
+            Ok(_) => panic!("unknown label must be refused"),
+        };
+        let msg = err2.to_string();
+        assert!(msg.starts_with("unknown fold 'p95'") && msg.contains("mean"), "{msg}");
+
+        assert!(!root.join("runs").exists(), "refusals fire before any store write");
+    }
+
+    /// C1: two identical record-all runs produce byte-identical tap files.
+    #[test]
+    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());
+        let a = std::fs::read(
+            root_a.join("runs").join("traces").join("sma_signal").join("fast_tap.json"),
+        )
+        .expect("run a trace");
+        let b = std::fs::read(
+            root_b.join("runs").join("traces").join("sma_signal").join("fast_tap.json"),
+        )
+        .expect("run b trace");
+        assert_eq!(a, b, "same input, same bytes (the index carries env provenance; the tap file must not)");
+    }
 }