diff --git a/crates/aura-cli/src/main.rs b/crates/aura-cli/src/main.rs index 5f93dc8..785115a 100644 --- a/crates/aura-cli/src/main.rs +++ b/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 ` 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 = BTreeSet::new(); + let mut tap_drains: Vec<(String, ScalarKind, mpsc::Receiver<(Timestamp, Vec)>)> = Vec::new(); + let declared: Vec = 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 = tap_drains + .into_iter() + .map(|(name, kind, rx)| { + let rows: Vec<(Timestamp, Vec)> = 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 } } diff --git a/crates/aura-cli/tests/tap_recording.rs b/crates/aura-cli/tests/tap_recording.rs new file mode 100644 index 0000000..c4fac4f --- /dev/null +++ b/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 ` 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 = (1..R_SMA_PRICES.len()) + .map(|i| (R_SMA_PRICES[i - 1] + R_SMA_PRICES[i]) / 2.0) + .collect(); + let expected_ts: Vec = (2..=R_SMA_PRICES.len() as i64).collect(); + + let got_ts: Vec = 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 = 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"); +}