From 34ff539143008e00a62e030ab29ee67ab2e95bb3 Mon Sep 17 00:00:00 2001 From: claude Date: Sun, 19 Jul 2026 16:09:51 +0200 Subject: [PATCH] =?UTF-8?q?feat:=20measurement=20run=20verb=20=E2=80=94=20?= =?UTF-8?q?shape-dispatch=20`aura=20run`=20on=20blueprint=20output?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase 3 of the Stratification milestone (contract C28). `aura run` now shape-dispatches on the loaded blueprint: a `bias` output takes the existing strategy path (wrap_r + R evaluation), byte-identical; a blueprint that declares taps but exposes no `bias` runs bare via the new `run_measurement` (bootstrap, run, drain the declared taps) with no SimBroker, no risk executor, and no per-cycle equity/exposure/r recorders; a blueprint exposing neither is refused. This gives the measurement layer its own run verb and removes the O(cycles) recording retention (the measured ~2 GB peak RSS over 12y UK100 M1) the mandatory R scaffold imposed on a measurement-shaped run. A new `MeasurementReport { manifest, taps }` (aura-engine::report, beside RunReport) is the stdout/record shape — no R metrics, since a bare run has no broker. The manifest's `broker` carries an interim "measurement" sentinel; the honest `Option` model is deferred to the #147 metric-genericity block, keeping this phase #147-free and the strategy-path C18 goldens byte-identical. `run_signal_r` is untouched (the tap machinery is duplicated into `run_measurement`, not extracted). The measurement arm's closed-blueprint guard uses `signal.param_space()` rather than `blueprint_axis_probe`, which welds wrap_r and would panic on a no-`bias` signal. refs #286 --- crates/aura-cli/src/main.rs | 197 ++++++++++++++++++----- crates/aura-cli/tests/run_measurement.rs | 58 +++++++ crates/aura-engine/src/lib.rs | 4 +- crates/aura-engine/src/report.rs | 17 ++ 4 files changed, 236 insertions(+), 40 deletions(-) create mode 100644 crates/aura-cli/tests/run_measurement.rs diff --git a/crates/aura-cli/src/main.rs b/crates/aura-cli/src/main.rs index f7a3ed9..670c64b 100644 --- a/crates/aura-cli/src/main.rs +++ b/crates/aura-cli/src/main.rs @@ -19,7 +19,8 @@ use aura_engine::{ blueprint_from_json, blueprint_to_json, f64_field, join_on_ts, monte_carlo, param_stability, r_metrics_from_rs, summarize, summarize_r, walk_forward, window_of, BindError, BlueprintNode, Composite, FamilySelection, GraphBuilder, Harness, JoinedRow, McAggregate, - McFamily, RBootstrap, RollMode, RunManifest, RunMetrics, RunReport, SelectionMode, + McFamily, MeasurementReport, RBootstrap, RollMode, RunManifest, RunMetrics, RunReport, + SelectionMode, SweepFamily, SweepPoint, SyntheticSpec, VecSource, WalkForwardResult, WindowBounds, WindowRoller, WindowRun, }; @@ -1812,6 +1813,106 @@ fn run_signal_r( RunReport { manifest, metrics } } +/// The measurement-run manifest: the honest subset of [`sim_optimal_manifest`] +/// with no broker/R specifics. `broker` carries the interim `"measurement"` +/// sentinel — `RunManifest.broker` is a required `String` today; the honest +/// `Option` model is deferred to the #147 metric-genericity block so this +/// phase stays #147-free and the strategy-path C18 goldens byte-identical. +fn measurement_manifest( + params: Vec<(String, Scalar)>, + window: (Timestamp, Timestamp), + seed: u64, +) -> RunManifest { + RunManifest { + commit: ENGINE_COMMIT.to_string(), + params, + defaults: Vec::new(), + window, + seed, + broker: "measurement".to_string(), + selection: None, + instrument: None, + topology_hash: None, + project: None, + } +} + +/// The bare measurement run (C28 phase 3): `run_signal_r` MINUS `wrap_r` and the +/// 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)] +fn run_measurement( + signal: Composite, params: &[Scalar], data: RunData, seed: u64, env: &project::Env, +) -> MeasurementReport { + let topo = topology_hash(&signal); + let run_name = signal.name().to_string(); + let binding = binding::resolve_binding(signal.name(), signal.input_roles(), &BTreeMap::new()) + .unwrap_or_else(|m| { + eprintln!("aura: {m}"); + std::process::exit(1); + }); + if matches!(data, RunData::Synthetic) && !binding.close_only() { + eprintln!("aura: {}", binding::synthetic_refusal(signal.name(), &binding)); + std::process::exit(1); + } + let names: Vec = signal.param_space().iter().map(|p| p.name.clone()).collect(); + let defaults = wrapped_bound_defaults(&signal); + let (sources, window, _pip_size) = resolve_run_data(&data, env, &binding); + + // Compile the signal DIRECTLY — no wrap_r, no broker/executor/eq-ex-r sinks. + let mut flat = signal.compile_with_params(params).unwrap_or_else(|e| { + eprintln!("aura: this blueprint does not compile to a runnable harness: {e:?}"); + std::process::exit(1); + }); + + // Bind each declared tap to a Recorder (mirrors run_signal_r:1759-1773). + 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 measurement harness"); + h.run_bound(key_supply(&binding, sources)) + .expect("sources opened against `binding` key-match that binding's own roles by construction"); + + let named_params: Vec<(String, Scalar)> = + names.into_iter().zip(params.iter().copied()).collect(); + let mut manifest = measurement_manifest(named_params, window, seed); + manifest.defaults = defaults; + manifest.topology_hash = Some(topo); + manifest.project = env.provenance(); + + // Drain + persist each declared tap (mirrors run_signal_r:1799-1811). + 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); + }); + } + MeasurementReport { manifest, taps: tap_names } +} + /// #260: the r-sma sugar/MC paths below run with either no cost model (empty /// `cost` slice) or CLI-flag cost specs (scalar-only — `cost_specs_from_params` /// only ever wraps `CostValue::Scalar`), so an instrument-keyed map can never @@ -3586,47 +3687,67 @@ fn dispatch_run(a: RunCmd, env: &project::Env) { eprintln!("aura: {path}: {msg}"); std::process::exit(2); }); - // `wrap_r` (below, via `run_signal_r` AND the axis probe right here) welds - // the loaded signal by its hardcoded `"bias"` output port (the R-evaluator - // contract). A signal exposing any other output shape (a measurement-only - // blueprint) would otherwise only surface as `blueprint_axis_probe`'s - // `wrap_r` call panicking on `BuildError::UnknownOutPort` a few lines - // down — refuse here instead, before that probe runs. - if !signal.output().iter().any(|o| o.name == "bias") { - let exposed: Vec<&str> = signal.output().iter().map(|o| o.name.as_str()).collect(); + // Shape dispatch (C28 phase 3): a `bias` output → the strategy path + // (wrap_r + R evaluation), byte-identical; else ≥1 declared tap → a + // bare measurement run (no wrap_r); else an inert blueprint → refuse. + // The closed-blueprint guard differs per arm: the strategy arm keeps + // `blueprint_axis_probe` (which welds wrap_r by the `bias` port); the + // measurement arm uses the signal's own `param_space()` (wrap_r-free — + // probing a no-`bias` signal would panic on UnknownOutPort). + let has_bias = signal.output().iter().any(|o| o.name == "bias"); + let has_tap = !signal.taps().is_empty(); + if has_bias { + // Refuse an open (free-knob) blueprint at the dispatch boundary, + // mirroring `blueprint_mc_family`'s closed-guard: `run` bootstraps + // over the EMPTY point, so a free knob would panic in + // `compile_with_params` — reject it clean instead (#176). + let free = blueprint_axis_probe(&doc, env).param_space(); + if !free.is_empty() { + eprintln!( + "aura: run requires a closed blueprint (no free parameters); {} free knob(s) — \ + bind them or use `aura sweep --axis`", + free.len() + ); + std::process::exit(2); + } + let params = match a.params.as_deref() { + Some(j) => parse_param_cells(j).unwrap_or_else(|m| { + eprintln!("aura: {m}"); + std::process::exit(2); + }), + None => Vec::new(), + }; + let data = run_data_from(a.real.as_deref(), a.from, a.to); + let report = run_signal_r(signal, ¶ms, data, a.seed.unwrap_or(0), env); + println!("{}", report.to_json()); + } else if has_tap { + // Measurement path: wrap_r-free closed guard via the signal's own + // open knobs (blueprint_axis_probe would weld wrap_r and panic). + if !signal.param_space().is_empty() { + eprintln!( + "aura: run requires a closed blueprint (no free parameters); {} free knob(s) — \ + bind them or use `aura sweep --axis`", + signal.param_space().len() + ); + std::process::exit(2); + } + let params = match a.params.as_deref() { + Some(j) => parse_param_cells(j).unwrap_or_else(|m| { + eprintln!("aura: {m}"); + std::process::exit(2); + }), + 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); + println!("{}", report.to_json()); + } else { eprintln!( - "aura: `aura run` expects a strategy with a bias output; this blueprint \ - exposes {}", - if exposed.is_empty() { - "no outputs".to_string() - } else { - exposed.join(", ") - } + "aura: `aura run` needs either a `bias` output (a strategy) or ≥1 \ + declared tap (a measurement); this blueprint exposes neither" ); std::process::exit(1); } - // Refuse an open (free-knob) blueprint at the dispatch boundary, mirroring - // `blueprint_mc_family`'s closed-guard: `run` bootstraps over the EMPTY point, so a - // free knob would panic in `compile_with_params` — reject it clean instead (#176). - let free = blueprint_axis_probe(&doc, env).param_space(); - if !free.is_empty() { - eprintln!( - "aura: run requires a closed blueprint (no free parameters); {} free knob(s) — \ - bind them or use `aura sweep --axis`", - free.len() - ); - std::process::exit(2); - } - let params = match a.params.as_deref() { - Some(j) => parse_param_cells(j).unwrap_or_else(|m| { - eprintln!("aura: {m}"); - std::process::exit(2); - }), - None => Vec::new(), - }; - let data = run_data_from(a.real.as_deref(), a.from, a.to); - let report = run_signal_r(signal, ¶ms, data, a.seed.unwrap_or(0), env); - println!("{}", report.to_json()); } None => { eprintln!( diff --git a/crates/aura-cli/tests/run_measurement.rs b/crates/aura-cli/tests/run_measurement.rs new file mode 100644 index 0000000..aa3de5e --- /dev/null +++ b/crates/aura-cli/tests/run_measurement.rs @@ -0,0 +1,58 @@ +//! #286 / C28 phase 3: `aura run` on a measurement-shaped blueprint (declares a +//! tap, exposes no `bias`) runs BARE — no wrap_r scaffold — emits a +//! `MeasurementReport` and persists the tapped series. A taps-only blueprint is +//! refused before this feature; this pins the new additive behaviour. + +use std::path::Path; +use std::process::Command; + +const BIN: &str = env!("CARGO_BIN_EXE_aura"); + +fn temp_cwd(name: &str) -> std::path::PathBuf { + let dir = Path::new(env!("CARGO_TARGET_TMPDIR")).join(format!("aura-cli-meas-{name}")); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).expect("create temp cwd"); + dir +} + +/// The shipped `examples/r_sma.json`, turned MEASUREMENT-shaped: one declared tap +/// on node 0 ("fast", SMA(2)) field 0, and its `bias` output removed. Same closed +/// topology (`sma_signal`), so it runs on the built-in synthetic stream. +fn measurement_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}}]); + v["blueprint"]["output"] = serde_json::json!([]); // no `bias` — a measurement, not a strategy + serde_json::to_string(&v).expect("re-serialize measurement blueprint") +} + +#[test] +fn measurement_blueprint_runs_bare_and_emits_its_tap() { + let cwd = temp_cwd("runs-bare"); + let bp_path = cwd.join("measurement.json"); + std::fs::write(&bp_path, measurement_blueprint_json()).expect("write measurement 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)); + + // stdout is a MeasurementReport: the declared tap name, no R `metrics`, the + // measurement broker sentinel. + let report: serde_json::Value = + serde_json::from_slice(&out.stdout).expect("stdout parses as MeasurementReport JSON"); + assert_eq!(report["taps"], serde_json::json!(["fast_tap"])); + assert!(report.get("metrics").is_none(), "a measurement run has no R metrics"); + assert_eq!(report["manifest"]["broker"], "measurement"); + + // The tapped series is persisted through the trace store. + 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"])); +} diff --git a/crates/aura-engine/src/lib.rs b/crates/aura-engine/src/lib.rs index f5fccb3..24fe441 100644 --- a/crates/aura-engine/src/lib.rs +++ b/crates/aura-engine/src/lib.rs @@ -71,8 +71,8 @@ pub use harness::{ pub use report::{ derive_position_events, expected_max_of_normals, f64_field, inv_norm_cdf, join_on_ts, r_metrics_from_rs, summarize, summarize_r, ColumnarTrace, FamilySelection, JoinedRow, - PositionAction, PositionEvent, ProjectProvenance, RMetrics, RunManifest, RunMetrics, - RunReport, SelectionMode, + MeasurementReport, PositionAction, PositionEvent, ProjectProvenance, RMetrics, RunManifest, + RunMetrics, RunReport, SelectionMode, }; pub use sweep::{ sweep, GridSpace, ListSpace, ParamRange, RandomSpace, Space, SweepError, SweepFamily, diff --git a/crates/aura-engine/src/report.rs b/crates/aura-engine/src/report.rs index bce1794..0504c42 100644 --- a/crates/aura-engine/src/report.rs +++ b/crates/aura-engine/src/report.rs @@ -111,6 +111,23 @@ impl RunReport { } } +/// A measurement run's stdout/record shape (C28 phase 3): the run descriptor plus +/// the declared tap names, mirroring the persisted `index.json`. A measurement run +/// has no broker → no equity/exposure → NO R `metrics` (the difference from +/// [`RunReport`]). The tapped series themselves persist through the trace store. +#[derive(Clone, Debug, PartialEq, serde::Serialize, serde::Deserialize)] +pub struct MeasurementReport { + pub manifest: RunManifest, + pub taps: Vec, +} + +impl MeasurementReport { + /// Canonical machine-readable JSON via serde, same encoder as [`RunReport`]. + pub fn to_json(&self) -> String { + serde_json::to_string(self).expect("a finite MeasurementReport always serializes") + } +} + /// Reduce a run's recorded pip-equity + exposure streams into summary metrics. /// Pure — identical inputs yield identical metrics (C1/C12). Timestamps are /// carried in the input to match exactly what a sink records; the reduction