feat(aura-cli): --tap TAP=FOLD fold-subscription selector on aura run

Decide #310's data-authorability fork as (b), split by run-mode
authority: selecting a Named fold subscription is data-reachable on the
one-shot run path; adding a fold stays a Rust entry (role 2, C25).
`aura run` gains a repeatable --tap TAP=FOLD selector feeding the
existing TapPlan seam: no flag keeps the record-all default; any flag
replaces the plan entirely (unlisted taps stay unbound — C27 inert,
non-error). Validation stays in bind_tap_plan (roster-enumerating label
refusal, typed UndeclaredTap, both before store I/O); parse-level
refusals (malformed pair, duplicate tap) exit 2 via the verb's usage
convention.

Rejected alternatives (minuted on #310 with the skeptic-pass
rationale): an op-script/blueprint carrier (C27 forbids recording
policy in the serialized fragment); ratifying Rust-only (softens a
designed-in promise with effort as the only rationale). The
campaign/document carrier and the persist_taps/declared-tap namespace
reconciliation are deferred to the Measurement-reachable milestone
(#312/#327, minuted on #312).

Ledger: C27 Current state records the boundary; the superseded sentence
moved verbatim to the new c27-declared-taps.history.md sidecar; the
glossary tap-plan paragraph updated, fixing the pre-existing inaccuracy
that named `aura measure` a record-all-passing verb (it is the post-hoc
IC analysis and constructs no tap plan).

Verification: 5 new unit tests (parse branches), 5 new binary-level
selector tests incl. a byte-identity pin of an explicit all-record plan
vs the record-all default (C1), 1 measurement-arm test; full workspace
suite green; clippy -D warnings clean. Independent opus diff review:
approved, no Important/Minor findings (nits: inert-arm generic refusal,
a=b=c falling to the roster refusal — accepted as-is).

closes #310

crates/aura-cli/src/main.rs

@@ -46,7 +46,7 @@ use aura_measurement::information_coefficient;
 // (a sibling module reaching it via `crate::blueprint_axis_probe_reopened`,
 // the crate-root re-export this `use` gives it), not from this module itself.
 use aura_runner::member::{blueprint_axis_probe, blueprint_axis_probe_reopened, run_signal_r, wrapped_bound_names, RunData};
-use aura_runner::TapPlan;
+use aura_runner::{TapPlan, TapSubscription};
 #[cfg(test)]
 use aura_runner::member::{run_blueprint_member, wrap_r, SYNTHETIC_PIP_SIZE};
 // The family builders (blueprint sweep / walk-forward / MC), the shared
@@ -105,6 +105,7 @@ use aura_std::Sma;
 #[cfg(test)]
 use std::sync::mpsc;
 use std::collections::HashSet;
+use std::collections::BTreeSet;
 #[cfg(test)]
 use std::collections::BTreeMap;
 use clap::{Args, Parser, Subcommand};
@@ -1343,6 +1344,12 @@ struct RunCmd {
     /// Not accepted — CLI-side trace persistence is retired (see #224).
     #[arg(long)]
     trace: Option<String>,
+    /// Subscribe a declared tap to a fold for this run (repeatable,
+    /// TAP=FOLD; e.g. --tap signal=mean). Replaces the record-all
+    /// default: only listed taps are bound, unlisted taps stay unbound.
+    /// Fold roster: `aura graph introspect --folds`.
+    #[arg(long = "tap", value_name = "TAP=FOLD")]
+    tap: Vec<String>,
 }
 
 #[derive(Args)]
@@ -1761,6 +1768,32 @@ fn campaign_window_ms(choice: DataChoice, env: &aura_runner::project::Env) -> (i
     )
 }
 
+/// Build the run's tap plan from repeated `--tap TAP=FOLD` selections
+/// (#310). No selections → the record-all default (today's behaviour).
+/// Any selection → an explicit plan that REPLACES the default entirely:
+/// only listed taps are bound; unlisted declared taps stay unbound,
+/// which C27 defines as inert, not an error. Tap/label existence is
+/// bind_tap_plan's to validate (roster refusal / UndeclaredTap), before
+/// any store I/O.
+fn tap_plan_from_args(args: &[String]) -> Result<TapPlan, String> {
+    if args.is_empty() {
+        return Ok(TapPlan::record_all());
+    }
+    let mut plan = TapPlan::empty();
+    let mut seen = BTreeSet::new();
+    for raw in args {
+        let (tap, label) = match raw.split_once('=') {
+            Some((t, l)) if !t.is_empty() && !l.is_empty() => (t, l),
+            _ => return Err(format!("--tap expects TAP=FOLD, got \"{raw}\"")),
+        };
+        if !seen.insert(tap.to_string()) {
+            return Err(format!("--tap names tap \"{tap}\" twice"));
+        }
+        plan.subscribe(tap, TapSubscription::named(label));
+    }
+    Ok(plan)
+}
+
 /// `aura run`: the loaded-blueprint branch (an existing `.json` first-positional) or
 /// the built-in harness-kind dispatch.
 fn dispatch_run(a: RunCmd, env: &aura_runner::project::Env) {
@@ -1797,6 +1830,13 @@ fn dispatch_run(a: RunCmd, env: &aura_runner::project::Env) {
             // 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();
+            let tap_plan = match tap_plan_from_args(&a.tap) {
+                Ok(p) => p,
+                Err(m) => {
+                    eprintln!("aura: {m}");
+                    std::process::exit(2);
+                }
+            };
             if has_bias {
                 // Refuse an open (free-knob) blueprint at the dispatch boundary,
                 // mirroring `blueprint_mc_family`'s closed-guard: `run` bootstraps
@@ -1819,7 +1859,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_signal_r(signal, &params, data, a.seed.unwrap_or(0), env, TapPlan::record_all());
+                let report = run_signal_r(signal, &params, data, a.seed.unwrap_or(0), env, tap_plan);
                 println!("{}", report.to_json());
             } else if has_tap {
                 // Measurement path: wrap_r-free closed guard via the signal's own
@@ -1840,7 +1880,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, &params, data, a.seed.unwrap_or(0), env, TapPlan::record_all());
+                let report = run_measurement(signal, &params, data, a.seed.unwrap_or(0), env, tap_plan);
                 println!("{}", report.to_json());
             } else {
                 eprintln!(
@@ -3859,5 +3899,41 @@ mod tests {
         assert_eq!(stop_length, R_SMA_STOP_LENGTH, "omitted --stop-length defaults to the regime");
         assert_eq!(stop_k, R_SMA_STOP_K, "omitted --stop-k defaults to the regime");
     }
+
+    #[test]
+    fn tap_plan_from_args_empty_is_record_all_ok() {
+        assert!(tap_plan_from_args(&[]).is_ok());
+    }
+
+    #[test]
+    fn tap_plan_from_args_refuses_a_pair_without_equals() {
+        let err = match tap_plan_from_args(&["fast_tapmean".to_string()]) {
+            Err(e) => e,
+            Ok(_) => panic!("expected an error"),
+        };
+        assert_eq!(err, "--tap expects TAP=FOLD, got \"fast_tapmean\"");
+    }
+
+    #[test]
+    fn tap_plan_from_args_refuses_empty_tap_or_label() {
+        assert!(tap_plan_from_args(&["=mean".to_string()]).is_err());
+        assert!(tap_plan_from_args(&["fast_tap=".to_string()]).is_err());
+    }
+
+    #[test]
+    fn tap_plan_from_args_refuses_the_same_tap_twice() {
+        let args = vec!["a=mean".to_string(), "a=last".to_string()];
+        let err = match tap_plan_from_args(&args) {
+            Err(e) => e,
+            Ok(_) => panic!("expected an error"),
+        };
+        assert_eq!(err, "--tap names tap \"a\" twice");
+    }
+
+    #[test]
+    fn tap_plan_from_args_accepts_distinct_selections() {
+        let args = vec!["a=mean".to_string(), "b=record".to_string()];
+        assert!(tap_plan_from_args(&args).is_ok());
+    }
 }
 

crates/aura-cli/tests/run_measurement.rs

@@ -56,3 +56,28 @@ fn measurement_blueprint_runs_bare_and_emits_its_tap() {
     assert_eq!(trace["tap"], "fast_tap");
     assert_eq!(trace["kinds"], serde_json::json!(["F64"]));
 }
+
+/// #310: the measurement arm honours the `--tap` selector — a `last`
+/// fold lands one summary row instead of the full series.
+#[test]
+fn measurement_run_honours_the_tap_selector() {
+    let cwd = temp_cwd("selector-last");
+    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"), "--tap", "fast_tap=last"])
+        .current_dir(&cwd)
+        .output()
+        .expect("spawn aura run");
+    assert!(out.status.success(), "stderr: {}", String::from_utf8_lossy(&out.stderr));
+
+    let trace_text = std::fs::read_to_string(cwd.join("runs/traces/sma_signal/fast_tap.json"))
+        .expect("read fold trace");
+    let trace: serde_json::Value = serde_json::from_str(&trace_text).expect("parse fold trace");
+    let rows = trace["columns"][0].as_array().expect("columns[0] array");
+    assert_eq!(rows.len(), 1, "a fold lands exactly one summary row");
+    // last warm SMA(2) value over the synthetic stream = (1.0097 + 1.0092) / 2
+    let got = rows[0].as_f64().expect("f64 row");
+    let expected = (1.0097_f64 + 1.0092) / 2.0;
+    assert!((got - expected).abs() < 1e-9, "last row: got {got}, expected {expected}");
+}

crates/aura-cli/tests/tap_recording.rs

@@ -124,6 +124,21 @@ fn duplicate_tap_blueprint_json() -> String {
     serde_json::to_string(&v).expect("re-serialize duplicate-tap blueprint")
 }
 
+/// The shipped `examples/r_sma.json` blueprint with two distinctly named
+/// declared taps: `fast_tap` on node 0 ("fast", SMA(2)) and `slow_tap` on
+/// node 1 ("slow") — the smallest fixture on which an explicit `--tap`
+/// plan and the record-all default can be compared file-by-file (#310).
+fn two_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}},
+        {"name": "slow_tap", "from": {"node": 1, "field": 0}},
+    ]);
+    serde_json::to_string(&v).expect("re-serialize two-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
@@ -212,3 +227,140 @@ fn a_read_only_run_dir_surfaces_terminally_through_the_tap_trace_register() {
     );
     assert!(!run_dir.join("index.json").exists(), "no index — treat-as-absent crash shape");
 }
+
+/// #310: `--tap fast_tap=mean` subscribes the declared tap to the `mean`
+/// fold instead of the record-all default — the trace store holds ONE
+/// summary row whose value is the mean of the full SMA(2) series, and
+/// `index.json` lists exactly the subscribed tap.
+#[test]
+fn run_tap_selector_persists_a_fold_summary_row() {
+    let cwd = temp_cwd("tap-selector-mean");
+    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"), "--tap", "fast_tap=mean"])
+        .current_dir(&cwd)
+        .output()
+        .expect("spawn aura run");
+    assert!(out.status.success(), "stderr: {}", String::from_utf8_lossy(&out.stderr));
+
+    let index_text = std::fs::read_to_string(cwd.join("runs/traces/sma_signal/index.json"))
+        .expect("read index.json");
+    let index: serde_json::Value = serde_json::from_str(&index_text).expect("parse index.json");
+    assert_eq!(index["taps"], serde_json::json!(["fast_tap"]));
+
+    let trace_text = std::fs::read_to_string(cwd.join("runs/traces/sma_signal/fast_tap.json"))
+        .expect("read fold trace");
+    let trace: serde_json::Value = serde_json::from_str(&trace_text).expect("parse fold trace");
+    assert_eq!(trace["tap"], "fast_tap");
+    let rows = trace["columns"][0].as_array().expect("columns[0] array");
+    assert_eq!(rows.len(), 1, "a fold lands exactly one summary row");
+    let sma: Vec<f64> = (1..R_SMA_PRICES.len())
+        .map(|i| (R_SMA_PRICES[i - 1] + R_SMA_PRICES[i]) / 2.0)
+        .collect();
+    let expected = sma.iter().sum::<f64>() / sma.len() as f64;
+    let got = rows[0].as_f64().expect("f64 row");
+    assert!((got - expected).abs() < 1e-9, "mean row: got {got}, expected {expected}");
+}
+
+/// #310: an all-`record` explicit plan is byte-identical to the no-flag
+/// record-all default — the selector replaces the default without
+/// changing what `record` itself writes (C1 determinism across runs).
+#[test]
+fn run_explicit_record_plan_matches_the_record_all_default() {
+    let cwd_a = temp_cwd("record-default");
+    let cwd_b = temp_cwd("record-explicit");
+    for cwd in [&cwd_a, &cwd_b] {
+        std::fs::write(cwd.join("two_tap.json"), two_tap_blueprint_json())
+            .expect("write two-tap blueprint");
+    }
+    let run = |cwd: &std::path::Path, extra: &[&str]| {
+        let mut args = vec!["run", "two_tap.json"];
+        args.extend_from_slice(extra);
+        let out = Command::new(BIN)
+            .args(&args)
+            .current_dir(cwd)
+            .output()
+            .expect("spawn aura run");
+        assert!(out.status.success(), "stderr: {}", String::from_utf8_lossy(&out.stderr));
+    };
+    run(&cwd_a, &[]);
+    run(&cwd_b, &["--tap", "fast_tap=record", "--tap", "slow_tap=record"]);
+
+    for file in ["fast_tap.json", "slow_tap.json"] {
+        let a = std::fs::read(cwd_a.join("runs/traces/sma_signal").join(file)).expect("default trace");
+        let b = std::fs::read(cwd_b.join("runs/traces/sma_signal").join(file)).expect("explicit trace");
+        assert_eq!(a, b, "{file} must be byte-identical across default and explicit record plans");
+    }
+    let taps_of = |cwd: &std::path::Path| -> serde_json::Value {
+        let text = std::fs::read_to_string(cwd.join("runs/traces/sma_signal/index.json"))
+            .expect("read index.json");
+        serde_json::from_str::<serde_json::Value>(&text).expect("parse index.json")["taps"].clone()
+    };
+    assert_eq!(taps_of(&cwd_a), taps_of(&cwd_b));
+}
+
+/// #310: an unknown fold label is refused through the registry's
+/// roster-enumerating refusal, before any store write.
+#[test]
+fn run_tap_selector_refuses_an_unknown_label_with_the_roster() {
+    let cwd = temp_cwd("tap-selector-unknown-label");
+    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"), "--tap", "fast_tap=medain"])
+        .current_dir(&cwd)
+        .output()
+        .expect("spawn aura run");
+    assert!(!out.status.success(), "an unknown fold label must refuse");
+    let stderr = String::from_utf8_lossy(&out.stderr);
+    assert!(
+        stderr.contains("medain") && stderr.contains("mean"),
+        "refusal must name the label and enumerate the roster: {stderr}"
+    );
+    assert!(!cwd.join("runs").exists(), "a refused run must not write the store");
+}
+
+/// #310: a selection naming a tap the blueprint does not declare is
+/// refused typed (UndeclaredTap), before any store write.
+#[test]
+fn run_tap_selector_refuses_an_undeclared_tap() {
+    let cwd = temp_cwd("tap-selector-undeclared");
+    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"), "--tap", "nope=mean"])
+        .current_dir(&cwd)
+        .output()
+        .expect("spawn aura run");
+    assert!(!out.status.success(), "an undeclared tap must refuse");
+    let stderr = String::from_utf8_lossy(&out.stderr);
+    assert!(stderr.contains("nope"), "refusal must name the offending tap: {stderr}");
+    assert!(!cwd.join("runs").exists(), "a refused run must not write the store");
+}
+
+/// #310: a malformed or duplicate `--tap` is a usage error (exit 2)
+/// before anything runs.
+#[test]
+fn run_tap_selector_refuses_malformed_and_duplicate_pairs() {
+    let cwd = temp_cwd("tap-selector-usage");
+    let bp_path = cwd.join("tapped_r_sma.json");
+    std::fs::write(&bp_path, tap_blueprint_json()).expect("write tapped blueprint");
+    for args in [
+        vec!["--tap", "fast_tapmean"],
+        vec!["--tap", "fast_tap=mean", "--tap", "fast_tap=last"],
+    ] {
+        let mut full = vec!["run", bp_path.to_str().expect("utf-8 path")];
+        full.extend(args);
+        let out = Command::new(BIN)
+            .args(&full)
+            .current_dir(&cwd)
+            .output()
+            .expect("spawn aura run");
+        assert_eq!(out.status.code(), Some(2), "usage errors exit 2");
+        let stderr = String::from_utf8_lossy(&out.stderr);
+        assert!(stderr.contains("--tap"), "usage message names the flag: {stderr}");
+        assert!(!cwd.join("runs").exists());
+    }
+}

docs/design/contracts/c27-declared-taps.history.md

@@ -0,0 +1,11 @@
+# C27 — Declared taps: named measurement points bind sinks run-mode-aware: history
+
+> FROZEN HISTORICAL RECORD. Each block below was true as of its cycle/date stamp
+> and may be superseded; this file is NOT current truth and NOT a grounding
+> surface. Current contract: [c27-declared-taps.md](c27-declared-taps.md).
+
+**Current-state tap-plan sentence (2026-07-21, #283; superseded 2026-07-24 by
+the #310 `--tap` selector):** "…and the shared `bind_tap_plan`/`BoundTaps`
+pair called by both declared-tap entry points, `run_signal_r`
+(`aura-runner::member`) and `run_measurement` (`aura-runner::measure`); both
+CLI verbs pass a record-all plan."

docs/design/contracts/c27-declared-taps.md

@@ -68,7 +68,18 @@ and the roster-enumerating refusal — plus a scalar-typed param schema; all cor
 entries are param-less today, the seam ships in every entry's build signature),
 and the shared `bind_tap_plan`/`BoundTaps` pair called by both declared-tap entry
 points, `run_signal_r` (`aura-runner::member`) and `run_measurement`
-(`aura-runner::measure`); both CLI verbs pass a record-all plan. The `record`
+(`aura-runner::measure`) — both arms of the single CLI verb `aura run`, whose
+repeatable `--tap TAP=FOLD` selector (#310) makes the `Named` selection
+data-reachable: no flag keeps the record-all default, any flag replaces the
+plan entirely (unlisted taps stay unbound/inert). The boundary is thereby
+fixed in place: *selecting* a subscription is run-mode authority, exercised
+by the run-mode owner — on the one-shot path the CLI invocation itself, a
+projection exercising this contract's authority, not a second home for
+intent under [C25](c25-role-model.md) — while *adding* a fold stays a Rust
+entry (role 2). The campaign/document carrier, and the reconciliation of the
+presentation-tap namespace (`persist_taps`) with declared taps it requires,
+is deferred to the Measurement-reachable milestone (#312/#327, minuted on
+#312). The `record`
 consumer (`aura-runner::tap_recorder::TapRecorder`) holds the trace store's
 streaming writer in-graph — `initialize` opens (deferred acquisition), `eval`
 appends `(Timestamp, Cell)` (zero per-cycle heap, #77), `finalize` reports

docs/glossary.md

@@ -327,7 +327,7 @@ A named recorded stream produced by a recording `sink` — the addressable label
 
 Since C27 (#282) the word also names a second, author-facing sense: a **declared tap** — a `Composite.taps` entry `Tap { name, from: {node, field} }` a hand-authored blueprint declares on an interior output wire, the output-side twin of an `input_role`. It is a pure declaration (no channel endpoint); the harness binds it run-mode-aware (a single `aura run` constructs a recorder at each and persists the series through the trace store; a sweep leaves it unbound and inert). This is an OPEN, per-blueprint author-declared name — distinct from the CLOSED campaign `persist_taps` vocabulary above, which selects among fixed observables of the standard R-harness. Both senses land as `ColumnarTrace`s in the same trace store; the closed vocabulary is what a `campaign document` selects, the declared tap is what a blueprint author names.
 
-Since #283 what CONSUMES a declared tap is itself declared per run by a **tap plan**: a subscription is either `Named { label, params }` — resolved against a layered **fold registry** whose core vocabulary is `record | count | sum | mean | min | max | first | last`, each entry carrying a doc line (the help surface and the roster-enumerating refusal) and a scalar-typed param schema (all core entries param-less; growth is a new Rust entry per C25, and higher layers register entries without touching the core) — or `Live(closure)`, the single non-data variant (an in-process consumer with consumer-owned loss policy). `record` streams the full series to the trace store at constant memory (no buffer-then-drain); folds keep an O(1) accumulator and land one summary row at finalize. Both CLI verbs (`aura run`, `aura measure`) pass a record-all plan, so their semantics are unchanged.
+Since #283 what CONSUMES a declared tap is itself declared per run by a **tap plan**: a subscription is either `Named { label, params }` — resolved against a layered **fold registry** whose core vocabulary is `record | count | sum | mean | min | max | first | last`, each entry carrying a doc line (the help surface and the roster-enumerating refusal) and a scalar-typed param schema (all core entries param-less; growth is a new Rust entry per C25, and higher layers register entries without touching the core) — or `Live(closure)`, the single non-data variant (an in-process consumer with consumer-owned loss policy). `record` streams the full series to the trace store at constant memory (no buffer-then-drain); folds keep an O(1) accumulator and land one summary row at finalize. Both declared-tap entry points are arms of the single verb `aura run` (`aura measure` is the post-hoc IC analysis over already-persisted traces and constructs no tap plan); `aura run` subscribes every declared tap to `record` by default, and its repeatable `--tap TAP=FOLD` selector (#310) replaces that default with an explicit plan — only listed taps are bound, unlisted taps stay unbound/inert.
 
 ### topology hash
 **Avoid:** —