feat: measurement run verb — shape-dispatch aura run on blueprint output

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<String>` 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
2026-07-19 16:09:51 +02:00
parent 1c49d5dce2
commit 34ff539143
4 changed files with 236 additions and 40 deletions
@@ -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<String>` 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<String> = 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<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 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<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);
+        });
+    }
+    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, &params, 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, &params, 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, &params, data, a.seed.unwrap_or(0), env);
-            println!("{}", report.to_json());
        }
        None => {
            eprintln!(
@@ -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"]));
+}
@@ -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,
@@ -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<String>,
+}
+
+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