diff --git a/crates/aura-cli/src/main.rs b/crates/aura-cli/src/main.rs index e24a859..1d243c8 100644 --- a/crates/aura-cli/src/main.rs +++ b/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, ¶ms, data, a.seed.unwrap_or(0), env); + let report = run_measurement(signal, ¶ms, data, a.seed.unwrap_or(0), env, TapPlan::record_all()); println!("{}", report.to_json()); } else { eprintln!( diff --git a/crates/aura-cli/tests/tap_recording.rs b/crates/aura-cli/tests/tap_recording.rs index c4fac4f..1b4dbd2 100644 --- a/crates/aura-cli/tests/tap_recording.rs +++ b/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"); +} diff --git a/crates/aura-runner/src/measure.rs b/crates/aura-runner/src/measure.rs index 9bcd38b..d196f8d 100644 --- a/crates/aura-runner/src/measure.rs +++ b/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 = 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: {}", 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 = tap_drains.iter().map(|(n, _, _)| n.clone()).collect(); - 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); - }); - } + // Close the tap plan (mirrors run_signal_r; nothing buffered, #283). + let tap_names: Vec = 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 = (1..R_SMA_PRICES.len()) + .map(|i| (R_SMA_PRICES[i - 1] + R_SMA_PRICES[i]) / 2.0) + .collect(); + let mean = sma.iter().sum::() / 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"); + } +} diff --git a/crates/aura-runner/src/member.rs b/crates/aura-runner/src/member.rs index 9a60f36..b17b2a1 100644 --- a/crates/aura-runner/src/member.rs +++ b/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 = + series["ts"].as_array().unwrap().iter().map(|v| v.as_i64().unwrap()).collect(); + let vals: Vec = + 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::() / 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)"); + } }