feat(cli): record declared taps on the single-run path (#282)

The payoff: a hand-authored blueprint's declared taps become recorded output
from one `aura run`. run_signal_r binds each tap (hoisted into flat.taps after
wrap_r nests the signal) to a fresh Recorder before bootstrap, drains the
series after the run, and persists each as a ColumnarTrace via the existing
env.trace_store() — the same trace surface the campaign path feeds, so the
existing chart tooling reads them. Duplicate tap names are refused (the
caller-owned DuplicateBind dedup) before anything persists. A tap-free run
writes no trace store — byte-identical to today. Sweep/reduce leaves taps
unbound (run_blueprint_member untouched). Covered end-to-end: a persisted tap
series, the tap-free no-write guard, and the duplicate-name refusal.

refs #282

crates/aura-cli/src/main.rs

@@ -54,17 +54,18 @@ use aura_std::Scale;
 // `Delay`/`Scale` above.
 #[cfg(test)]
 use aura_std::{Add, Gt, Latch, Mul, Sqrt};
-// `Bias`/`Ema`/`Sma`/`ColumnarTrace` are only reached from the test module; the
-// imports are test-only, mirroring `Delay`/`Scale` above.
-#[cfg(test)]
+// `ColumnarTrace` is now also reached from `run_signal_r`'s tap-persistence
+// path, so it is a production import (no longer test-only).
 use aura_engine::ColumnarTrace;
+// `Bias`/`Ema`/`Sma` are only reached from the test module; the imports are
+// test-only, mirroring `Delay`/`Scale` above.
 #[cfg(test)]
 use aura_std::{Bias, Ema, Sma};
 #[cfg(test)]
 use aura_std::{ConstantCost, VolSlippageCost};
 use std::sync::mpsc;
 use std::sync::LazyLock;
-use std::collections::{BTreeMap, HashSet};
+use std::collections::{BTreeMap, BTreeSet, HashSet};
 use clap::{Args, Parser, Subcommand};
 
 /// The pip size the built-in *synthetic* families (sweep/walkforward/mc over a
@@ -1716,10 +1717,12 @@ fn resolve_run_data(
 /// 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.
+#[allow(clippy::type_complexity)]
 fn run_signal_r(
     signal: Composite, params: &[Scalar], data: RunData, seed: u64, env: &project::Env,
 ) -> RunReport {
     let topo = topology_hash(&signal); // 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.
     let binding = binding::resolve_binding(signal.name(), signal.input_roles(), &BTreeMap::new())
@@ -1745,9 +1748,28 @@ fn run_signal_r(
     let (tx_req, _rx_req) = mpsc::channel();
     let (sources, window, pip_size) = resolve_run_data(&data, env, &binding);
     let wrapped = wrap_r(signal, tx_eq, tx_ex, tx_r, tx_req, StopRule::Vol { length: R_SMA_STOP_LENGTH, k: R_SMA_STOP_K }, false, pip_size, &binding, None);
-    let flat = wrapped
+    let mut flat = wrapped
         .compile_with_params(params)
         .expect("signal binds + wraps to a valid harness");
+    // Bind each declared tap to a fresh `Recorder` + channel, before bootstrap
+    // — `flat.taps` already carries the signal's declared taps hoisted to the
+    // root. Dedup is the caller's per `TapBindError::DuplicateBind`'s doc (the
+    // engine keeps no cross-call state).
+    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: {}", aura_engine::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));
+    }
     let mut h = Harness::bootstrap(flat).expect("valid r-sma harness");
     // `sources` were opened via `resolve_run_data(&data, env, &binding)` against
     // this SAME `binding`, and `key_supply` keys them by that binding's own role
@@ -1769,6 +1791,23 @@ fn run_signal_r(
     manifest.project = env.provenance();
     let mut metrics = summarize(&f64_field(&eq_rows, 0), &f64_field(&ex_rows, 0));
     metrics.r = Some(summarize_r(&r_rows, &[]));
+    // Drain + persist each declared tap's series, guarded so a tap-free
+    // `aura run` stays byte-identical to today (no `runs/` write at all).
+    // `manifest` is built above so the persisted `index.json` carries this
+    // run's own provenance.
+    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);
+        });
+    }
     RunReport { manifest, metrics }
 }
 

crates/aura-cli/tests/tap_recording.rs

@@ -0,0 +1,152 @@
+//! #282: a hand-authored blueprint with a declared tap, run on the single-run
+//! path, records the tapped series; the same blueprint in a sweep does not
+//! (`run_blueprint_member`, the sweep/reduce path, never calls `bind_tap` —
+//! structurally guaranteed by Task 6 leaving it untouched, not re-asserted
+//! here).
+
+use std::path::Path;
+use std::process::Command;
+
+/// Path to the freshly-built `aura` binary (Cargo sets this env var for the
+/// test crate; the binary is named `aura` in `Cargo.toml`).
+const BIN: &str = env!("CARGO_BIN_EXE_aura");
+
+/// A fresh, unique working directory for this process test (so the run's
+/// `./runs/traces/` never dirties the repo). Unique per test + per process.
+fn temp_cwd(name: &str) -> std::path::PathBuf {
+    let dir = Path::new(env!("CARGO_TARGET_TMPDIR")).join(format!("aura-cli-tap-{name}"));
+    let _ = std::fs::remove_dir_all(&dir);
+    std::fs::create_dir_all(&dir).expect("create temp cwd");
+    dir
+}
+
+/// The exact synthetic price array `r_sma_prices()` (main.rs) feeds the
+/// `RunData::Synthetic` path — duplicated here (not re-exported; a private
+/// CLI helper) so the tapped SMA(2) column can be pinned against an
+/// independently-derived expectation, not a blind capture.
+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,
+];
+
+/// The shipped `examples/r_sma.json` blueprint, with one additional declared
+/// tap on node 0 ("fast", `SMA(length=2)`) field 0 — the smallest addition
+/// that names an interior producer output without touching the existing
+/// topology (nodes/edges/input_roles/output all verbatim).
+fn tap_blueprint_json() -> String {
+    let path = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
+    let doc = std::fs::read_to_string(path).expect("read 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}}]);
+    serde_json::to_string(&v).expect("re-serialize tapped blueprint")
+}
+
+/// Property: on the single-run path (`aura run <blueprint.json>` over the
+/// built-in synthetic stream, no `--real`), a declared tap is bound and its
+/// per-cycle series is persisted through the trace store — the persisted
+/// `fast_tap.json` carries exactly the SMA(2) values over the known synthetic
+/// price stream (silent until warm at sample 2, one value per cycle after).
+#[test]
+fn single_run_persists_a_declared_tap_series() {
+    let cwd = temp_cwd("single-run-persists");
+    let bp_path = cwd.join("tapped_r_sma.json");
+    std::fs::write(&bp_path, tap_blueprint_json()).expect("write tapped blueprint");
+
+    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(), "stderr: {}", String::from_utf8_lossy(&out.stderr));
+
+    let trace_path = cwd.join("runs/traces/sma_signal/fast_tap.json");
+    let trace_text = std::fs::read_to_string(&trace_path).unwrap_or_else(|e| {
+        panic!("expected a persisted tap trace at {}: {e}", trace_path.display())
+    });
+    let trace: serde_json::Value = serde_json::from_str(&trace_text).expect("parse tap trace json");
+
+    assert_eq!(trace["tap"], "fast_tap");
+    assert_eq!(trace["kinds"], serde_json::json!(["F64"]));
+
+    // SMA(2), silent until warm (2 samples): first emission at the 2nd sample
+    // (ts=2), value = mean(prices[0], prices[1]); then a rolling mean of the
+    // last two prices per subsequent cycle (ts = i+1 for prices[i]).
+    let expected: Vec<f64> = (1..R_SMA_PRICES.len())
+        .map(|i| (R_SMA_PRICES[i - 1] + R_SMA_PRICES[i]) / 2.0)
+        .collect();
+    let expected_ts: Vec<i64> = (2..=R_SMA_PRICES.len() as i64).collect();
+
+    let got_ts: Vec<i64> = trace["ts"].as_array().expect("ts array").iter().map(|v| v.as_i64().unwrap()).collect();
+    assert_eq!(got_ts, expected_ts, "recorded timestamps");
+
+    let got: Vec<f64> = trace["columns"][0]
+        .as_array()
+        .expect("columns[0] array")
+        .iter()
+        .map(|v| v.as_f64().unwrap())
+        .collect();
+    assert_eq!(got.len(), expected.len(), "recorded row count");
+    for (i, (g, e)) in got.iter().zip(expected.iter()).enumerate() {
+        assert!((g - e).abs() < 1e-9, "row {i}: got {g}, expected {e}");
+    }
+}
+
+/// A tap-free blueprint run (the existing `run_prints_json_and_exits_zero`
+/// shape) leaves no `runs/` directory at all — the tap machinery is fully
+/// inert (no binds, no trace_store write) when a blueprint declares no taps,
+/// matching Task 6's byte-identical guarantee.
+#[test]
+fn tap_free_run_writes_no_trace_store() {
+    let cwd = temp_cwd("tap-free-writes-nothing");
+    let bp = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
+    let out = Command::new(BIN)
+        .args(["run", &bp])
+        .current_dir(&cwd)
+        .output()
+        .expect("spawn aura run");
+    assert!(out.status.success(), "stderr: {}", String::from_utf8_lossy(&out.stderr));
+    assert!(!cwd.join("runs").exists(), "a tap-free run must not create a trace store");
+}
+
+/// The shipped `examples/r_sma.json` blueprint with TWO declared taps that
+/// share the same name (`"dup"`), on two distinct producers (node 0 "fast" and
+/// node 1 "slow") — the shape `run_signal_r`'s `seen: BTreeSet` dedup guard
+/// exists for (`FlatGraph::bind_tap` itself keeps no cross-call state per its
+/// own doc comment, so nothing upstream of the CLI catches this).
+fn duplicate_tap_blueprint_json() -> String {
+    let path = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
+    let doc = std::fs::read_to_string(path).expect("read 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": "dup", "from": {"node": 0, "field": 0}},
+        {"name": "dup", "from": {"node": 1, "field": 0}},
+    ]);
+    serde_json::to_string(&v).expect("re-serialize duplicate-tap blueprint")
+}
+
+/// Property: **a blueprint declaring two taps under the same name is refused
+/// on the single-run path with a named error and exit code 1, before any
+/// trace is persisted** — the CLI-owned `TapBindError::DuplicateBind` dedup
+/// guard (bind_tap itself binds a single tap and keeps no cross-call state,
+/// per its own doc comment) actually fires end-to-end, and the partial first
+/// bind never reaches disk as a half-written trace store.
+#[test]
+fn single_run_refuses_a_duplicate_tap_name_before_persisting_anything() {
+    let cwd = temp_cwd("duplicate-tap-name-refused");
+    let bp_path = cwd.join("dup_tap_r_sma.json");
+    std::fs::write(&bp_path, duplicate_tap_blueprint_json()).expect("write dup-tap blueprint");
+
+    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(), "a duplicate tap name must be refused, not silently bound");
+    let stderr = String::from_utf8_lossy(&out.stderr);
+    assert!(
+        stderr.contains("dup") && stderr.contains("more than once"),
+        "stderr must name the offending tap and the duplicate-bind fault: {stderr}"
+    );
+    assert!(!cwd.join("runs").exists(), "a refused run must not persist a half-written trace store");
+}