fieldtest: tap-subscribers — 5 examples, 0 bugs / 2 friction / 3 spec-gap

Cycle-close fieldtest for #283 + #77 (tap subscribers), exercising the new
tap surface as a downstream consumer over the public interface only (design
ledger, glossary, cargo doc, example corpus — never the crate sources).

Five examples, one per axis, all built and run against worktree HEAD bd0c557:
- tap_1_record_spread.ops.json    — record → runs/traces/graph/spread.json
- tap_4_measure_ic.ops.json       — two-tap run → aura measure ic (clean IC)
- tap_consumer/src/tap_2_fold.rs  — Named{mean}/Named{count} via run_signal_r
- tap_consumer/src/tap_3_live.rs  — Live(closure); bit-identical to fold + record
- tap_consumer/examples/refusal_probe.rs — unknown-label refusal (exit 1, no write)

The consumer crate is a detached workspace (own empty [workspace], path-deps on
aura-core/-engine/-runner), exactly as a real World program; it is not a
workspace member and does not build under `cargo build --workspace`. Verified
here with a fresh `cargo build --bins --examples` (Finished, 0 errors).

No merge-blocking defect: record/fold/live/measure over the shared pair are
correct and deterministic. Follow-ups filed at cycle close:
- #309 (feature) — a plain `aura run` echoes no recorded trace handle (F1+SG1)
- #310 (idea)    — Named fold has no data/CLI reach; ratify Rust-only vs selector (SG2)
- #311 (bug)     — same-name re-run orphans still-chartable tap files (SG3, pre-existing)
- #297 (comment) — library refusal = process::exit, not Result (F2, already tracked)

crates/aura-cli/src/main.rs

@@ -45,6 +45,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;
 #[cfg(test)]
 use aura_runner::member::{run_blueprint_member, wrap_r, SYNTHETIC_PIP_SIZE};
 // The family builders (blueprint sweep / walk-forward / MC), the shared
@@ -1765,7 +1766,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);
+                let report = run_signal_r(signal, &params, data, a.seed.unwrap_or(0), env, TapPlan::record_all());
                 println!("{}", report.to_json());
             } else if has_tap {
                 // Measurement path: wrap_r-free closed guard via the signal's own
@@ -1786,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, &params, data, a.seed.unwrap_or(0), env);
+                let report = run_measurement(signal, &params, data, a.seed.unwrap_or(0), env, TapPlan::record_all());
                 println!("{}", report.to_json());
             } else {
                 eprintln!(
@@ -3099,6 +3100,7 @@ mod tests {
             RunData::Synthetic,
             0,
             &env,
+            TapPlan::record_all(),
         );
         let member4 = &family.points[0].report; // slow=4 is the first odometer point
         assert_eq!(member4.metrics, single.metrics, "loaded sweep member == single run");

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");
+}

crates/aura-core/src/node.rs

@@ -379,6 +379,15 @@ pub trait Node {
         "node".to_string()
     }
 
+    /// Start-of-stream hook: `Harness::run` calls it once per node, in
+    /// topological order, before the first source value — the mirror of
+    /// [`Node::finalize`]. A consumer overrides it to acquire run resources
+    /// (e.g. the file a recording sink streams into). Infallible by
+    /// signature: an implementation that can fail stores its error, degrades
+    /// to inert, and surfaces the failure once, terminally, at `finalize`.
+    /// The default is a no-op, so existing nodes are unaffected.
+    fn initialize(&mut self) {}
+
     /// End-of-stream hook: `Harness::run` calls it once per node, in topological
     /// order, after the source loop drains. A folding sink overrides it to flush
     /// its accumulated summary; the default is a no-op, so existing nodes are

crates/aura-engine/src/harness.rs

@@ -519,6 +519,14 @@ impl Harness {
         let mut cycle_id: u64 = 0;
         let mut scratch: Vec<Cell> = Vec::new();
 
+        // start-of-stream: initialize every node once, in topological order,
+        // before the first source value (the mirror of the end-of-stream
+        // finalize flush below). Runs inside the deterministic sequence — no
+        // within-sim concurrency (C1).
+        for &nidx in topo.iter() {
+            nodes[nidx].node.initialize();
+        }
+
         loop {
             // pick the live source head with the smallest (timestamp, source index).
             // strictly-`<` replace + source-order scan preserves C4 tie-breaking.
@@ -2809,6 +2817,52 @@ mod tests {
         assert_eq!(fired, vec![0, 1]);
     }
 
+    struct InitProbe {
+        id: usize,
+        tx: mpsc::Sender<String>,
+        evaled: bool,
+    }
+    impl Node for InitProbe {
+        fn lookbacks(&self) -> Vec<usize> {
+            vec![1]
+        }
+        fn initialize(&mut self) {
+            let _ = self.tx.send(format!("init{}", self.id));
+        }
+        fn eval(&mut self, _ctx: Ctx<'_>) -> Option<&[Cell]> {
+            if !self.evaled {
+                self.evaled = true;
+                let _ = self.tx.send(format!("eval{}", self.id));
+            }
+            None
+        }
+    }
+
+    #[test]
+    fn run_initializes_every_node_once_before_the_stream() {
+        let (tx, rx) = mpsc::channel();
+        // two sink probes, both fed by the single source; no inter-node edges.
+        let mut h = boot(
+            vec![
+                Box::new(InitProbe { id: 0, tx: tx.clone(), evaled: false }),
+                Box::new(InitProbe { id: 1, tx, evaled: false }),
+            ],
+            vec![
+                recorder_sig(&[ScalarKind::F64], Firing::Any),
+                recorder_sig(&[ScalarKind::F64], Firing::Any),
+            ],
+            vec![SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }])],
+            vec![],
+        )
+        .expect("valid");
+        // nothing initialized before the run.
+        assert!(rx.try_recv().is_err());
+        h.run(vec![Box::new(VecSource::new(f64_stream(&[(1, 1.0), (2, 2.0)])))]);
+        // both inits fire, in topo order, strictly before any eval.
+        let fired: Vec<String> = rx.try_iter().collect();
+        assert_eq!(fired, vec!["init0", "init1", "eval0", "eval1"]);
+    }
+
     #[test]
     fn bind_tap_attaches_a_caller_built_sink_and_records_the_tapped_series() {
         // A tap on a producer; bind_tap attaches a caller-built Recorder; the run

crates/aura-registry/src/lib.rs

@@ -37,7 +37,10 @@ pub use lineage::{
 };
 
 mod trace_store;
-pub use trace_store::{FamilyMember, NameKind, RunTraces, TraceStore, TraceStoreError, WriteKind};
+pub use trace_store::{
+    FamilyMember, NameKind, RunTraces, TapTraceWriter, TapWriterOpener, TraceStore,
+    TraceStoreError, TraceStreamer, WriteKind,
+};
 
 /// An append-only run registry over a JSONL file: one serde_json line per
 /// `RunReport`.

crates/aura-registry/src/trace_store.rs

@@ -7,8 +7,11 @@
 
 use std::fmt;
 use std::fs;
+use std::fs::{File, OpenOptions};
+use std::io::{self, BufWriter, Seek, SeekFrom, Write};
 use std::path::{Path, PathBuf};
 
+use aura_core::{Cell, ScalarKind, Timestamp};
 use aura_engine::{ColumnarTrace, RunManifest};
 use serde::{Deserialize, Serialize};
 
@@ -97,24 +100,19 @@ impl TraceStore {
         let run_dir = self.dir.join(name);
         fs::create_dir_all(&run_dir)?;
         for tap in taps {
-            let path = run_dir.join(format!("{}.json", tap.tap));
-            let body = serde_json::to_string(tap).map_err(|source| TraceStoreError::Parse {
-                file: path.display().to_string(),
-                source,
-            })?;
-            fs::write(&path, body)?;
+            write_tap_file(&run_dir, tap)?;
         }
-        let index = Index {
-            manifest: manifest.clone(),
-            taps: taps.iter().map(|t| t.tap.clone()).collect(),
-        };
-        let index_path = run_dir.join("index.json");
-        let body = serde_json::to_string(&index).map_err(|source| TraceStoreError::Parse {
-            file: index_path.display().to_string(),
-            source,
-        })?;
-        fs::write(&index_path, body)?;
-        Ok(())
+        let order: Vec<String> = taps.iter().map(|t| t.tap.clone()).collect();
+        write_index(&run_dir, manifest, &order)
+    }
+
+    /// Begin a streamed run write under `traces/<name>/` (directories created
+    /// now; `index.json` only at [`TraceStreamer::finish`] — the crash
+    /// discipline `read` already honors: no index → `NotFound`).
+    pub fn begin_run(&self, name: &str) -> Result<TraceStreamer, TraceStoreError> {
+        let run_dir = self.dir.join(name);
+        fs::create_dir_all(&run_dir)?;
+        Ok(TraceStreamer { run_dir })
     }
 
     /// Read a run's index + taps back, in index order. A missing run directory (no
@@ -239,6 +237,216 @@ impl TraceStore {
     }
 }
 
+/// One tap's canonical `<tap>.json` under `run_dir` — the single write shape
+/// both `TraceStore::write` and `TraceStreamer::write_full` share.
+fn write_tap_file(run_dir: &Path, tap: &ColumnarTrace) -> Result<(), TraceStoreError> {
+    let path = run_dir.join(format!("{}.json", tap.tap));
+    let body = serde_json::to_string(tap).map_err(|source| TraceStoreError::Parse {
+        file: path.display().to_string(),
+        source,
+    })?;
+    fs::write(&path, body)?;
+    Ok(())
+}
+
+/// The run's `index.json` (manifest + tap order) — shared by `write` and
+/// `TraceStreamer::finish`. Written last by both paths (crash discipline:
+/// no `index.json` → `NotFound`, treat-as-absent).
+fn write_index(
+    run_dir: &Path,
+    manifest: &RunManifest,
+    taps: &[String],
+) -> Result<(), TraceStoreError> {
+    let index = Index { manifest: manifest.clone(), taps: taps.to_vec() };
+    let index_path = run_dir.join("index.json");
+    let body = serde_json::to_string(&index).map_err(|source| TraceStoreError::Parse {
+        file: index_path.display().to_string(),
+        source,
+    })?;
+    fs::write(&index_path, body)?;
+    Ok(())
+}
+
+/// A begun streamed run write: hands out per-tap incremental writers and
+/// writes the closing `index.json`. Tap order in the index is the caller's
+/// (declared-tap order — the caller tracks it; this type keeps no order
+/// state so record-writer threads and post-run fold writes can interleave).
+pub struct TraceStreamer {
+    run_dir: PathBuf,
+}
+
+impl TraceStreamer {
+    /// An incremental writer for one tap, opened now. `Send`. Expressed
+    /// through [`TraceStreamer::register_tap`] so there is exactly one open
+    /// path and the two cannot drift.
+    pub fn tap_writer(&self, tap: &str, kind: ScalarKind) -> Result<TapTraceWriter, TraceStoreError> {
+        self.register_tap(tap, kind).open()
+    }
+
+    /// The deferred twin of [`TraceStreamer::tap_writer`]: everything needed
+    /// to open one tap's incremental writer later — from inside a consumer
+    /// node's `initialize`, at run start — without holding the streamer.
+    /// `Send`; opens no handles until [`TapWriterOpener::open`].
+    pub fn register_tap(&self, tap: &str, kind: ScalarKind) -> TapWriterOpener {
+        TapWriterOpener { run_dir: self.run_dir.clone(), tap: tap.to_string(), kind }
+    }
+
+    /// Write one complete tap file at once (the fold path's one-row trace) —
+    /// the same bytes `TraceStore::write` produces for it.
+    pub fn write_full(&self, tap: &ColumnarTrace) -> Result<(), TraceStoreError> {
+        write_tap_file(&self.run_dir, tap)
+    }
+
+    /// Close the run: write `index.json` (manifest + `taps` in caller order).
+    pub fn finish(self, manifest: &RunManifest, taps: &[String]) -> Result<(), TraceStoreError> {
+        write_index(&self.run_dir, manifest, taps)
+    }
+}
+
+/// See [`TraceStreamer::register_tap`].
+pub struct TapWriterOpener {
+    run_dir: PathBuf,
+    tap: String,
+    kind: ScalarKind,
+}
+
+impl TapWriterOpener {
+    /// Open the two temp column streams and return the incremental writer.
+    pub fn open(self) -> Result<TapTraceWriter, TraceStoreError> {
+        let final_path = self.run_dir.join(format!("{}.json", self.tap));
+        let ts_path = self.run_dir.join(format!("{}.ts.tmp", self.tap));
+        let col_path = self.run_dir.join(format!("{}.col.tmp", self.tap));
+        // Read+write handles, NOT `File::create`: `finish` seeks these same
+        // handles back and `io::copy`-reads them, and a `File::create` handle
+        // is O_WRONLY — reading it back fails with EBADF (os error 9).
+        let ts_w = BufWriter::new(
+            OpenOptions::new().read(true).write(true).create(true).truncate(true).open(&ts_path)?,
+        );
+        let col_w = BufWriter::new(
+            OpenOptions::new().read(true).write(true).create(true).truncate(true).open(&col_path)?,
+        );
+        Ok(TapTraceWriter {
+            tap: self.tap,
+            kind: self.kind,
+            final_path,
+            ts_path,
+            col_path,
+            ts_w,
+            col_w,
+            any: false,
+        })
+    }
+}
+
+/// The four-kind tag string (the on-disk `ColumnarTrace.kinds` form). A local
+/// copy of `report.rs`'s private `kind_tag`; the streamed-vs-write
+/// byte-equality test below is the drift guard.
+fn tag(kind: ScalarKind) -> &'static str {
+    match kind {
+        ScalarKind::F64 => "F64",
+        ScalarKind::I64 => "I64",
+        ScalarKind::Bool => "Bool",
+        ScalarKind::Timestamp => "Timestamp",
+    }
+}
+
+/// Coerce one cell to the f64 a trace column stores — the same table as
+/// `ColumnarTrace::from_rows` (`scalar_to_f64`): f64 as-is, i64 as f64,
+/// bool 1.0/0.0, timestamp epoch as f64.
+fn cell_to_f64(kind: ScalarKind, cell: Cell) -> f64 {
+    match kind {
+        ScalarKind::F64 => cell.f64(),
+        ScalarKind::I64 => cell.i64() as f64,
+        ScalarKind::Bool => {
+            if cell.bool() {
+                1.0
+            } else {
+                0.0
+            }
+        }
+        ScalarKind::Timestamp => cell.ts().0 as f64,
+    }
+}
+
+/// Incremental single-column trace writer: appends each `(ts, cell)` to two
+/// temp streams, then `finish` assembles the canonical `ColumnarTrace` JSON
+/// by streaming concatenation (constant memory) and removes the temps. The
+/// output is byte-equal to `serde_json::to_string(&ColumnarTrace)` for the
+/// same rows — pinned by test against `TraceStore::write`, which is the
+/// anchor (not a frozen byte-shape assumption).
+pub struct TapTraceWriter {
+    tap: String,
+    kind: ScalarKind,
+    final_path: PathBuf,
+    ts_path: PathBuf,
+    col_path: PathBuf,
+    ts_w: BufWriter<File>,
+    col_w: BufWriter<File>,
+    any: bool,
+}
+
+impl TapTraceWriter {
+    /// Append one recorded point. Number formatting is `serde_json`'s own
+    /// (`to_writer` per value) — identical to the whole-struct serialize.
+    pub fn append(&mut self, ts: Timestamp, cell: Cell) -> Result<(), TraceStoreError> {
+        if self.any {
+            self.ts_w.write_all(b",")?;
+            self.col_w.write_all(b",")?;
+        }
+        serde_json::to_writer(&mut self.ts_w, &ts.0).map_err(|source| TraceStoreError::Parse {
+            file: self.ts_path.display().to_string(),
+            source,
+        })?;
+        serde_json::to_writer(&mut self.col_w, &cell_to_f64(self.kind, cell)).map_err(
+            |source| TraceStoreError::Parse {
+                file: self.col_path.display().to_string(),
+                source,
+            },
+        )?;
+        self.any = true;
+        Ok(())
+    }
+
+    /// Assemble `{"tap":…,"kinds":[…],"ts":[⟨ts-tmp⟩],"columns":[[⟨col-tmp⟩]]}`
+    /// and remove the temps.
+    pub fn finish(self) -> Result<(), TraceStoreError> {
+        let TapTraceWriter { tap, kind, final_path, ts_path, col_path, ts_w, col_w, any: _ } = self;
+        let mut ts_f = ts_w.into_inner().map_err(|e| TraceStoreError::Io(e.into_error()))?;
+        let mut col_f = col_w.into_inner().map_err(|e| TraceStoreError::Io(e.into_error()))?;
+        ts_f.flush()?;
+        col_f.flush()?;
+        ts_f.seek(SeekFrom::Start(0))?;
+        col_f.seek(SeekFrom::Start(0))?;
+
+        let mut out = BufWriter::new(File::create(&final_path)?);
+        // Field order (tap, kinds, ts, columns) = ColumnarTrace declaration
+        // order; string fields serialized via serde for correct escaping.
+        write!(
+            out,
+            "{{\"tap\":{},\"kinds\":[{}],\"ts\":[",
+            serde_json::to_string(&tap).map_err(|source| TraceStoreError::Parse {
+                file: final_path.display().to_string(),
+                source,
+            })?,
+            serde_json::to_string(tag(kind)).map_err(|source| TraceStoreError::Parse {
+                file: final_path.display().to_string(),
+                source,
+            })?,
+        )?;
+        io::copy(&mut ts_f, &mut out)?;
+        out.write_all(b"],\"columns\":[[")?;
+        io::copy(&mut col_f, &mut out)?;
+        out.write_all(b"]]}")?;
+        out.flush()?;
+        drop(out);
+        drop(ts_f);
+        drop(col_f);
+        fs::remove_file(&ts_path)?;
+        fs::remove_file(&col_path)?;
+        Ok(())
+    }
+}
+
 /// What can go wrong reading or writing the trace store.
 #[derive(Debug)]
 pub enum TraceStoreError {
@@ -447,4 +655,168 @@ mod tests {
         assert!(store.ensure_name_free("fresh", WriteKind::Run).is_ok());
         let _ = fs::remove_dir_all(&root);
     }
+
+    /// The load-bearing compatibility pin: for the same rows, the streamed
+    /// writer's file bytes equal `TraceStore::write`'s — the legacy path's
+    /// bytes, whatever they are, define the shape. Covers all four
+    /// `ScalarKind` arms (f64 fractional/negative, bool, i64, timestamp) —
+    /// every arm `tag`/`cell_to_f64` dispatch on, since those two helpers are
+    /// a local copy of `report.rs`'s private `kind_tag`/`scalar_to_f64` and
+    /// this is the only drift guard against them — plus the empty
+    /// (never-warm) tap.
+    #[test]
+    fn streamed_writer_is_byte_equal_to_write_for_the_same_rows() {
+        use aura_core::Cell;
+
+        type Case<'a> = (&'a str, ScalarKind, Vec<(Timestamp, Cell)>);
+        let cases: Vec<Case> = vec![
+            (
+                "f64_tap",
+                ScalarKind::F64,
+                vec![
+                    (Timestamp(2), Cell::from_f64(0.30000000000000004)),
+                    (Timestamp(3), Cell::from_f64(-1.5)),
+                    (Timestamp(5), Cell::from_f64(2.0)),
+                ],
+            ),
+            (
+                "bool_tap",
+                ScalarKind::Bool,
+                vec![(Timestamp(1), Cell::from_bool(true)), (Timestamp(2), Cell::from_bool(false))],
+            ),
+            (
+                "i64_tap",
+                ScalarKind::I64,
+                vec![(Timestamp(1), Cell::from_i64(-7)), (Timestamp(2), Cell::from_i64(42))],
+            ),
+            (
+                "timestamp_tap",
+                ScalarKind::Timestamp,
+                vec![
+                    (Timestamp(1), Cell::from_ts(Timestamp(100))),
+                    (Timestamp(2), Cell::from_ts(Timestamp(-3))),
+                ],
+            ),
+            ("empty_tap", ScalarKind::F64, vec![]),
+        ];
+
+        for (tap, kind, rows) in cases {
+            // legacy path
+            let legacy_root = temp_traces_root(&format!("byteeq-legacy-{tap}"));
+            let legacy = TraceStore::open(&legacy_root);
+            let scalar_rows: Vec<(Timestamp, Vec<Scalar>)> = rows
+                .iter()
+                .map(|&(t, c)| (t, vec![Scalar::from_cell(kind, c)]))
+                .collect();
+            let trace = ColumnarTrace::from_rows(tap, &[kind], &scalar_rows);
+            legacy.write("run", &sample_manifest(), &[trace]).expect("legacy write");
+            let legacy_bytes =
+                fs::read(legacy_root.join(format!("traces/run/{tap}.json"))).expect("legacy bytes");
+
+            // streamed path
+            let stream_root = temp_traces_root(&format!("byteeq-stream-{tap}"));
+            let store = TraceStore::open(&stream_root);
+            let streamer = store.begin_run("run").expect("begin_run");
+            let mut w = streamer.tap_writer(tap, kind).expect("tap_writer");
+            for (t, c) in rows {
+                w.append(t, c).expect("append");
+            }
+            w.finish().expect("finish");
+            let streamed_bytes =
+                fs::read(stream_root.join(format!("traces/run/{tap}.json"))).expect("streamed bytes");
+
+            assert_eq!(
+                streamed_bytes, legacy_bytes,
+                "streamed and legacy write bytes must be identical for {tap}"
+            );
+            // temps removed
+            assert!(!stream_root.join(format!("traces/run/{tap}.ts.tmp")).exists());
+            assert!(!stream_root.join(format!("traces/run/{tap}.col.tmp")).exists());
+
+            let _ = fs::remove_dir_all(&legacy_root);
+            let _ = fs::remove_dir_all(&stream_root);
+        }
+    }
+
+    /// A begun-but-unfinished streamed run (crash shape: tap file written or
+    /// half-written, no `index.json`) reads as `NotFound` — treat-as-absent.
+    #[test]
+    fn begun_but_unfinished_streamed_run_reads_not_found() {
+        use aura_core::Cell;
+        let root = temp_traces_root("stream-crash");
+        let store = TraceStore::open(&root);
+        let streamer = store.begin_run("crashed").expect("begin_run");
+        let mut w = streamer.tap_writer("t", ScalarKind::F64).expect("tap_writer");
+        w.append(Timestamp(1), Cell::from_f64(1.0)).expect("append");
+        // no w.finish(), no streamer.finish(): the crash shape.
+        drop(w);
+        match store.read("crashed") {
+            Err(TraceStoreError::NotFound(name)) => assert_eq!(name, "crashed"),
+            other => panic!("expected NotFound, got {other:?}"),
+        }
+        let _ = fs::remove_dir_all(&root);
+    }
+
+    /// `register_tap`/`TapWriterOpener::open` produce the exact same bytes as
+    /// `tap_writer` for the same rows — the deferred opener and the
+    /// open-now writer are one write path (`tap_writer` delegates to
+    /// `register_tap(..).open()`), so they cannot drift.
+    #[test]
+    fn register_tap_open_is_the_same_write_path_as_tap_writer() {
+        use aura_core::Cell;
+        let root = temp_traces_root("deferred-open");
+        let store = TraceStore::open(&root);
+        let rows =
+            [(Timestamp(2), Cell::from_f64(0.25)), (Timestamp(3), Cell::from_f64(-1.5))];
+
+        // run "a": open-now.
+        let sa = store.begin_run("a").expect("begin a");
+        let mut wa = sa.tap_writer("t", ScalarKind::F64).expect("open now");
+        for &(ts, c) in &rows {
+            wa.append(ts, c).expect("append a");
+        }
+        wa.finish().expect("finish a");
+
+        // run "b": deferred — the opener is Send and opens at use time.
+        let sb = store.begin_run("b").expect("begin b");
+        let opener = sb.register_tap("t", ScalarKind::F64);
+        fn assert_send<T: Send>(_: &T) {}
+        assert_send(&opener);
+        let mut wb = opener.open().expect("deferred open");
+        for &(ts, c) in &rows {
+            wb.append(ts, c).expect("append b");
+        }
+        wb.finish().expect("finish b");
+
+        let a = fs::read(root.join("traces").join("a").join("t.json")).expect("read a");
+        let b = fs::read(root.join("traces").join("b").join("t.json")).expect("read b");
+        assert_eq!(a, b, "one open path, identical bytes");
+    }
+
+    /// `TraceStreamer::finish` + `write_full` produce a run `read` returns in
+    /// caller order — the fold path's one-row trace beside a streamed tap.
+    #[test]
+    fn streamer_finish_indexes_streamed_and_full_taps_in_caller_order() {
+        use aura_core::Cell;
+        let root = temp_traces_root("stream-index");
+        let store = TraceStore::open(&root);
+        let streamer = store.begin_run("mixed").expect("begin_run");
+        let mut w = streamer.tap_writer("series", ScalarKind::F64).expect("tap_writer");
+        w.append(Timestamp(1), Cell::from_f64(1.0)).expect("append");
+        w.finish().expect("finish series");
+        let fold_row = ColumnarTrace::from_rows(
+            "agg",
+            &[ScalarKind::I64],
+            &[(Timestamp(1), vec![Scalar::i64(1)])],
+        );
+        streamer.write_full(&fold_row).expect("write_full");
+        streamer
+            .finish(&sample_manifest(), &["series".to_string(), "agg".to_string()])
+            .expect("streamer finish");
+
+        let back = store.read("mixed").expect("read");
+        let names: Vec<&str> = back.taps.iter().map(|t| t.tap.as_str()).collect();
+        assert_eq!(names, vec!["series", "agg"], "index order = caller order");
+        let _ = fs::remove_dir_all(&root);
+    }
 }

crates/aura-runner/src/lib.rs

@@ -18,10 +18,17 @@ pub mod member;
 pub mod project;
 pub mod reproduce;
 pub mod runner;
+pub mod tap_plan;
+pub mod tap_recorder;
 pub mod translate;
 
 pub use project::Env;
 pub use runner::DefaultMemberRunner;
+pub use tap_plan::{
+    bind_tap_plan, BoundTaps, FoldBuildCtx, FoldEntry, FoldOutput, FoldRegistry, FoldSink, TapPlan,
+    TapPlanError, TapSubscription,
+};
+pub use tap_recorder::TapRecorder;
 
 /// A refusal a library function reports instead of exiting the process
 /// itself. The shell (`aura-cli`'s `dispatch_reproduce`) is the single place

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<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: {}", 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<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);
-        });
-    }
+    // Close the tap plan (mirrors run_signal_r; nothing buffered, #283).
+    let tap_names: Vec<String> = 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<f64> = (1..R_SMA_PRICES.len())
+            .map(|i| (R_SMA_PRICES[i - 1] + R_SMA_PRICES[i]) / 2.0)
+            .collect();
+        let mean = sma.iter().sum::<f64>() / 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");
+    }
+}

crates/aura-runner/src/member.rs

@@ -9,14 +9,14 @@
 //! `CliMemberRunner`, and any downstream World program) drive real data
 //! through.
 
-use std::collections::{BTreeMap, BTreeSet, HashSet};
+use std::collections::{BTreeMap, HashSet};
 use std::sync::{mpsc, Arc, LazyLock};
 
 use aura_composites::{cost_graph, risk_executor, StopRule};
 use aura_core::{zip_params, Cell, Firing, ParamSpec, PrimitiveBuilder, Scalar, ScalarKind, Timestamp};
 use aura_engine::{
-    blueprint_from_json, window_of, BlueprintNode, ColumnarTrace, Composite, GraphBuilder, Harness,
-    RunManifest, VecSource,
+    blueprint_from_json, window_of, BlueprintNode, Composite, GraphBuilder, Harness, RunManifest,
+    VecSource,
 };
 use aura_backtest::{
     summarize, summarize_r, RunMetrics, RunReport, SimBroker, PM_FIELD_NAMES, PM_RECORD_KINDS,
@@ -26,6 +26,7 @@ use aura_strategy::{cost_port, GEOMETRY_WIDTH};
 
 use crate::binding::ResolvedBinding;
 use crate::project::Env;
+use crate::tap_plan::{bind_tap_plan, TapPlan};
 use crate::translate::{R_SMA_STOP_LENGTH, R_SMA_STOP_K};
 
 /// The single build-time commit provenance (`option_env!("AURA_COMMIT")`,
@@ -480,6 +481,7 @@ pub fn resolve_run_data(
 #[allow(clippy::type_complexity)]
 pub fn run_signal_r(
     signal: Composite, params: &[Scalar], data: RunData, seed: u64, env: &Env,
+    plan: TapPlan,
 ) -> RunReport {
     // topology_hash's own two-line body, inlined: `content_id_of` over the
     // canonical (#164) blueprint JSON — the CLI shell's `topology_hash`
@@ -517,25 +519,14 @@ pub fn run_signal_r(
         eprintln!("aura: this blueprint does not compile to a runnable harness: {e:?}");
         std::process::exit(1);
     });
-    // 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<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));
-    }
+    // Bind each declared tap per the plan's subscription, before bootstrap
+    // (#283): typed refusals for bad plans, record consumers hold their
+    // streaming writer in-graph, folds accumulate O(1), live closures run
+    // inline. Dedup stays caller-owned per TapBindError::DuplicateBind's doc.
+    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 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
@@ -557,23 +548,13 @@ pub fn run_signal_r(
     manifest.project = env.provenance();
     let mut metrics = summarize(&aura_engine::f64_field(&eq_rows, 0), &aura_engine::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<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);
-        });
-    }
+    // Close the tap plan: drain the ≤1-message channels, write fold rows,
+    // then `index.json` — nothing was buffered during the run (#283). A
+    // tap-free (or nothing-persisting) plan wrote nothing at all.
+    bound.finish(&manifest).unwrap_or_else(|e| {
+        eprintln!("aura: {e}");
+        std::process::exit(1);
+    });
     RunReport { manifest, metrics }
 }
 
@@ -954,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 —
@@ -1158,8 +1142,8 @@ mod tests {
         let env = Env::std();
         let loaded = blueprint_from_json(include_str!("../../aura-cli/examples/r_breakout.json"), &|t| std_vocabulary(t))
             .expect("shipped r_breakout example loads");
-        let via_file = run_signal_r(loaded, &[], RunData::Synthetic, 0, &env);
-        let via_carve = run_signal_r(r_breakout_signal(Some(R_BREAKOUT_CHANNEL)), &[], RunData::Synthetic, 0, &env);
+        let via_file = run_signal_r(loaded, &[], RunData::Synthetic, 0, &env, TapPlan::record_all());
+        let via_carve = run_signal_r(r_breakout_signal(Some(R_BREAKOUT_CHANNEL)), &[], RunData::Synthetic, 0, &env, TapPlan::record_all());
         assert_eq!(via_file.metrics, via_carve.metrics, "loaded example grades identically to the carve");
     }
 
@@ -1311,13 +1295,14 @@ mod tests {
         let env = Env::std();
         let loaded = blueprint_from_json(include_str!("../../aura-cli/examples/r_meanrev.json"), &|t| std_vocabulary(t))
             .expect("shipped r_meanrev example loads");
-        let via_file = run_signal_r(loaded, &[], RunData::Synthetic, 0, &env);
+        let via_file = run_signal_r(loaded, &[], RunData::Synthetic, 0, &env, TapPlan::record_all());
         let via_carve = run_signal_r(
             r_meanrev_signal(Some(R_MEANREV_WINDOW), Some(R_MEANREV_BAND_K)),
             &[],
             RunData::Synthetic,
             0,
             &env,
+            TapPlan::record_all(),
         );
         assert_eq!(via_file.metrics, via_carve.metrics, "loaded example grades identically to the carve");
     }
@@ -1646,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<i64> =
+            series["ts"].as_array().unwrap().iter().map(|v| v.as_i64().unwrap()).collect();
+        let vals: Vec<f64> =
+            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::<f64>() / 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)");
+    }
 }

crates/aura-runner/src/tap_plan.rs

@@ -0,0 +1,639 @@
+//! The per-run tap subscription plan (#283, 2026-07-21 design revision) and
+//! the layered fold registry it resolves against. A tap stays a pure
+//! declaration (C27); this module is where the run/measurement side declares
+//! what consumes it. Record, fold, and live are ONE mechanism — a consumer
+//! of the tap's `(Timestamp, Cell)` stream; the only real axis is data-layer
+//! serializability: a `Named` subscription (label + scalar params) is fully
+//! expressible as data, a `Live` closure is the single deliberately non-data
+//! variant. The registry mirrors the node-vocabulary pattern
+//! (`std_vocabulary` + `Env::resolve`): a closed, typed vocabulary whose
+//! growth is a new Rust entry (C25) — core seeds here, higher layers
+//! (e.g. trading) register their own entries without bleeding into the core.
+
+use std::collections::{BTreeMap, BTreeSet};
+use std::fmt;
+use std::sync::mpsc::{self, Receiver, Sender};
+
+use aura_core::{Cell, Firing, Node, NodeSchema, ParamSpec, PortSpec, Scalar, ScalarKind, Timestamp};
+use aura_engine::{ColumnarTrace, FlatGraph, RunManifest, TapBindError};
+use aura_registry::{TapWriterOpener, TraceStoreError, TraceStreamer};
+use aura_std::{fold_binds_at, fold_output_kind, FoldKind, TapFold, TapLive};
+
+use crate::project::Env;
+use crate::tap_recorder::TapRecorder;
+
+/// What consumes one declared tap during a run — the drain policy, declared
+/// on the run/measurement side (#283), never on the tap.
+pub enum TapSubscription {
+    /// A consumer from the plan's fold registry, by label, with scalar-typed
+    /// construction parameters (v1 core entries take none).
+    Named { label: String, params: Vec<(String, Scalar)> },
+    /// An in-process consumer; loss policy is the closure's own.
+    Live(Box<dyn FnMut(Timestamp, Cell) + Send>),
+}
+
+impl TapSubscription {
+    /// A named subscription with no construction parameters.
+    pub fn named(label: &str) -> Self {
+        TapSubscription::Named { label: label.to_string(), params: Vec::new() }
+    }
+
+    /// A named subscription with construction parameters.
+    pub fn named_with(label: &str, params: Vec<(String, Scalar)>) -> Self {
+        TapSubscription::Named { label: label.to_string(), params }
+    }
+
+    /// Sugar for a live subscription around a closure.
+    pub fn live(consumer: impl FnMut(Timestamp, Cell) + Send + 'static) -> Self {
+        TapSubscription::Live(Box::new(consumer))
+    }
+}
+
+/// Per-run tap plan: name → subscription; unnamed taps get the default. The
+/// plan carries the registry it resolves against (the `Env`-at-resolve
+/// pattern), so a layered host injects its extended vocabulary as data. The
+/// only defaults are `named("record")` (`record_all` — the one entry that
+/// binds at every kind and takes no params) and `None` (empty plan; unnamed
+/// taps stay unbound, inert per C27) — fold/live subscriptions are per-tap,
+/// never a default, structurally (the field is private and no constructor
+/// sets another).
+pub struct TapPlan {
+    pub(crate) registry: FoldRegistry,
+    pub(crate) default_named: Option<(String, Vec<(String, Scalar)>)>,
+    pub(crate) by_name: BTreeMap<String, TapSubscription>,
+}
+
+impl TapPlan {
+    /// Record every declared tap — the CLI's plan on both verbs.
+    pub fn record_all() -> Self {
+        Self::record_all_with(FoldRegistry::core())
+    }
+
+    /// `record_all` over an injected (layered) registry.
+    pub fn record_all_with(registry: FoldRegistry) -> Self {
+        TapPlan {
+            registry,
+            default_named: Some(("record".to_string(), Vec::new())),
+            by_name: BTreeMap::new(),
+        }
+    }
+
+    /// No default: only explicitly subscribed taps are consumed.
+    pub fn empty() -> Self {
+        TapPlan { registry: FoldRegistry::core(), default_named: None, by_name: BTreeMap::new() }
+    }
+
+    /// Subscribe one tap by name (replaces an earlier subscription for it).
+    pub fn subscribe(&mut self, tap: &str, sub: TapSubscription) {
+        self.by_name.insert(tap.to_string(), sub);
+    }
+}
+
+/// Persistence shape of an entry's consumer.
+pub enum FoldOutput {
+    /// The full series, streamed (the `record` entry).
+    Series,
+    /// One summary row at finalize, of this kind.
+    Row(ScalarKind),
+}
+
+/// The sink a build call receives — matching the entry's declared
+/// [`FoldOutput`]: `Row` entries get the one-row channel, `Series` entries
+/// the deferred writer opener + terminal-outcome channel.
+pub enum FoldSink {
+    Row(Sender<(Timestamp, Vec<Scalar>)>),
+    Series(TapWriterOpener, Sender<Result<(), TraceStoreError>>),
+}
+
+/// What the wiring hands a registry entry's `build`.
+pub struct FoldBuildCtx {
+    /// The tap's column kind (already `binds_at`-validated).
+    pub kind: ScalarKind,
+    /// Construction parameters, already validated against the entry's schema.
+    pub params: Vec<(String, Scalar)>,
+    pub sink: FoldSink,
+}
+
+/// One registry entry: a labelled consumer constructor with docs, a param
+/// schema, bind rules, and its persistence shape.
+pub struct FoldEntry {
+    pub label: &'static str,
+    /// One line for help generation and roster-enumerating refusals.
+    pub doc: &'static str,
+    /// Scalar-typed construction parameters (the node vocabulary's param
+    /// plane). All v1 core entries: empty — the seam ships now because it
+    /// sits in every entry's build signature (C25: a param that would carry
+    /// logic is a new entry, never a freetext hole).
+    pub params: Vec<ParamSpec>,
+    /// May this entry bind a tap column of this kind?
+    pub binds_at: Box<dyn Fn(ScalarKind) -> bool>,
+    /// What the consumer persists, given the tap's column kind.
+    pub output: Box<dyn Fn(ScalarKind) -> FoldOutput>,
+    /// Build the consumer node. `ctx.params` arrive validated; `ctx.sink`
+    /// matches `output` (the wiring consults `output` first — an entry may
+    /// `panic!` on the wrong variant as a wiring-contract violation).
+    pub build: Box<dyn Fn(FoldBuildCtx) -> Box<dyn Node>>,
+}
+
+/// The label → entry map. Layered: `core()` seeds the standard vocabulary;
+/// `register` adds (or deliberately shadows) an entry — dependency
+/// injection, no core edit.
+pub struct FoldRegistry {
+    entries: BTreeMap<&'static str, FoldEntry>,
+}
+
+impl FoldRegistry {
+    /// The core vocabulary: `record`, `count`, `sum`, `mean`, `min`, `max`,
+    /// `first`, `last`.
+    pub fn core() -> Self {
+        let mut r = FoldRegistry { entries: BTreeMap::new() };
+        r.register(FoldEntry {
+            label: "record",
+            doc: "persist the full series, lossless, at constant memory (any kind)",
+            params: Vec::new(),
+            binds_at: Box::new(|_| true),
+            output: Box::new(|_| FoldOutput::Series),
+            build: Box::new(|ctx| {
+                let FoldSink::Series(opener, outcome_tx) = ctx.sink else {
+                    panic!("wiring contract: record is a Series entry");
+                };
+                Box::new(TapRecorder::new(ctx.kind, opener, outcome_tx))
+            }),
+        });
+        for (label, doc, fold) in [
+            ("count", "number of warm rows (any kind; i64 row)", FoldKind::Count),
+            ("sum", "sum of the series (f64 taps; f64 row)", FoldKind::Sum),
+            ("mean", "arithmetic mean of the series (f64 taps; f64 row)", FoldKind::Mean),
+            ("min", "minimum of the series (f64 taps; f64 row)", FoldKind::Min),
+            ("max", "maximum of the series (f64 taps; f64 row)", FoldKind::Max),
+            (
+                "first",
+                "first warm value, at its own timestamp (any kind; kind-preserving row)",
+                FoldKind::First,
+            ),
+            ("last", "last warm value (any kind; kind-preserving row)", FoldKind::Last),
+        ] {
+            r.register(FoldEntry {
+                label,
+                doc,
+                params: Vec::new(),
+                binds_at: Box::new(move |k| fold_binds_at(fold, k)),
+                output: Box::new(move |k| FoldOutput::Row(fold_output_kind(fold, k))),
+                build: Box::new(move |ctx| {
+                    let FoldSink::Row(tx) = ctx.sink else {
+                        panic!("wiring contract: {fold:?} is a Row entry");
+                    };
+                    Box::new(TapFold::new(ctx.kind, fold, tx))
+                }),
+            });
+        }
+        r
+    }
+
+    /// Add an entry. A later registration under an existing label replaces
+    /// it (layers may deliberately shadow).
+    pub fn register(&mut self, entry: FoldEntry) {
+        self.entries.insert(entry.label, entry);
+    }
+
+    pub fn get(&self, label: &str) -> Option<&FoldEntry> {
+        self.entries.get(label)
+    }
+
+    /// `(label, doc)` in label order — the help surface and the refusal
+    /// roster.
+    pub fn roster(&self) -> Vec<(&'static str, &'static str)> {
+        self.entries.values().map(|e| (e.label, e.doc)).collect()
+    }
+
+    /// The labels alone, label order (refusal messages).
+    pub fn labels(&self) -> Vec<&'static str> {
+        self.entries.keys().copied().collect()
+    }
+}
+
+/// A typed tap-plan fault — the pre-bootstrap refusals plus the terminal
+/// store fault. Entry points map every variant to the established
+/// `aura: ` + exit-1 refusal register via `Display`.
+pub enum TapPlanError {
+    /// The plan names a tap the blueprint does not declare.
+    UnknownTap { name: String },
+    /// The plan names a label the registry does not carry.
+    UnknownLabel { label: String, roster: Vec<&'static str> },
+    /// The entry's bind rule rejects the tap's column kind.
+    KindMismatch { tap: String, label: String, kind: ScalarKind },
+    /// A param binding names no schema param (`takes` = the schema's names).
+    UnknownParam { label: String, name: String, takes: Vec<String> },
+    /// A schema param is unbound.
+    MissingParam { label: String, name: String, kind: ScalarKind },
+    /// A param binding's kind mismatches its schema declaration.
+    ParamKind { label: String, name: String, expected: ScalarKind, got: ScalarKind },
+    /// Two declared taps share a name (the caller-owned dedup guard).
+    Bind(aura_engine::TapBindError),
+    /// The trace store failed (begin/write/finish) — the terminal register.
+    Store(TraceStoreError),
+}
+
+impl fmt::Display for TapPlanError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            TapPlanError::UnknownTap { name } => {
+                write!(f, "the tap plan names '{name}', but the blueprint declares no such tap")
+            }
+            TapPlanError::UnknownLabel { label, roster } => {
+                write!(f, "unknown fold '{label}' — available: {}", roster.join(", "))
+            }
+            TapPlanError::KindMismatch { tap, label, kind } => {
+                write!(f, "fold '{label}' cannot bind tap '{tap}' of kind {kind:?}")
+            }
+            TapPlanError::UnknownParam { label, name, takes } => {
+                if takes.is_empty() {
+                    write!(f, "fold '{label}' takes no parameters (got '{name}')")
+                } else {
+                    write!(f, "fold '{label}' has no parameter '{name}' — takes: {}", takes.join(", "))
+                }
+            }
+            TapPlanError::MissingParam { label, name, kind } => {
+                write!(f, "fold '{label}' requires parameter '{name}' ({kind:?})")
+            }
+            TapPlanError::ParamKind { label, name, expected, got } => {
+                write!(f, "fold '{label}' parameter '{name}' is {expected:?}, got {got:?}")
+            }
+            TapPlanError::Bind(e) => write!(f, "{e}"),
+            TapPlanError::Store(e) => write!(f, "writing tap traces failed: {e}"),
+        }
+    }
+}
+
+impl fmt::Debug for TapPlanError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "TapPlanError({self})")
+    }
+}
+
+impl From<TraceStoreError> for TapPlanError {
+    fn from(e: TraceStoreError) -> Self {
+        TapPlanError::Store(e)
+    }
+}
+
+/// Validate one `Named` subscription's param bindings against the entry's
+/// schema: every binding names a schema param of the right kind; every
+/// schema param is bound.
+pub(crate) fn validate_params(
+    entry: &FoldEntry,
+    bound: &[(String, Scalar)],
+) -> Result<(), TapPlanError> {
+    let takes: Vec<String> = entry.params.iter().map(|p| p.name.clone()).collect();
+    for (name, value) in bound {
+        match entry.params.iter().find(|p| &p.name == name) {
+            None => {
+                return Err(TapPlanError::UnknownParam {
+                    label: entry.label.to_string(),
+                    name: name.clone(),
+                    takes,
+                })
+            }
+            Some(p) if p.kind != value.kind() => {
+                return Err(TapPlanError::ParamKind {
+                    label: entry.label.to_string(),
+                    name: name.clone(),
+                    expected: p.kind,
+                    got: value.kind(),
+                })
+            }
+            Some(_) => {}
+        }
+    }
+    for p in &entry.params {
+        if !bound.iter().any(|(n, _)| n == &p.name) {
+            return Err(TapPlanError::MissingParam {
+                label: entry.label.to_string(),
+                name: p.name.clone(),
+                kind: p.kind,
+            });
+        }
+    }
+    Ok(())
+}
+
+/// The one-input sink schema every tap consumer binds with (empty output =
+/// pure consumer, C8; the port name is a non-load-bearing debug symbol).
+fn tap_sink_schema(kind: ScalarKind) -> NodeSchema {
+    NodeSchema {
+        inputs: vec![PortSpec { kind, firing: Firing::Any, name: "in".to_string() }],
+        output: vec![],
+        params: vec![],
+    }
+}
+
+/// The bound half of a tap plan: what `bind_tap_plan` leaves for the caller
+/// to drain after the run. Both declared-tap entry points call the pair, so
+/// the mechanism cannot drift between them.
+#[allow(clippy::type_complexity)]
+pub struct BoundTaps {
+    streamer: Option<TraceStreamer>,
+    declared: Vec<String>,
+    /// Persisting taps (Series and Row alike), declared-tap order — the
+    /// `index.json` tap order.
+    persisted: Vec<String>,
+    rows: Vec<(String, ScalarKind, Receiver<(Timestamp, Vec<Scalar>)>)>,
+    outcomes: Vec<(String, Receiver<Result<(), TraceStoreError>>)>,
+}
+
+impl BoundTaps {
+    /// Every declared tap, declared order (the measurement report's roster).
+    pub fn declared_names(&self) -> &[String] {
+        &self.declared
+    }
+
+    /// Drain the ≤1-message-per-tap channels and close the run: record
+    /// outcomes first (declared order), then fold rows written through the
+    /// same streamer, then `index.json` last. Any fault returns before the
+    /// index is written — the store's crash discipline (no `index.json` →
+    /// treat-as-absent) covers the partial directory, exactly like today's
+    /// failed `write`. A plan that persisted nothing began no streamer and
+    /// writes nothing.
+    pub fn finish(mut self, manifest: &RunManifest) -> Result<(), TapPlanError> {
+        for (_, rx) in &self.outcomes {
+            // finalize ran inside the harness's end-of-stream flush, so the
+            // single outcome is already in the channel.
+            if let Ok(Err(e)) = rx.try_recv() {
+                return Err(TapPlanError::Store(e));
+            }
+        }
+        let streamer = self.streamer.take();
+        for (name, kind, rx) in self.rows {
+            let rows: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect(); // ≤1
+            let trace = ColumnarTrace::from_rows(&name, &[kind], &rows);
+            streamer
+                .as_ref()
+                .expect("a Row tap implies a begun streamer")
+                .write_full(&trace)?;
+        }
+        if let Some(s) = streamer {
+            s.finish(manifest, &self.persisted)?;
+        }
+        Ok(())
+    }
+}
+
+/// Validate the plan against `flat.taps`, begin the streamed run write when
+/// anything persists, and bind one consumer per subscribed tap — the shared
+/// pre-bootstrap half of both declared-tap entry points. Refusals are typed
+/// and complete BEFORE any store I/O (a refused run never half-writes).
+pub fn bind_tap_plan(
+    flat: &mut FlatGraph,
+    mut plan: TapPlan,
+    env: &Env,
+    run_name: &str,
+) -> Result<BoundTaps, TapPlanError> {
+    let declared_taps: Vec<aura_engine::FlatTap> = flat.taps.clone();
+
+    // Dedup guard (caller-owned per TapBindError::DuplicateBind's doc).
+    let mut seen: BTreeSet<String> = BTreeSet::new();
+    for tap in &declared_taps {
+        if !seen.insert(tap.name.clone()) {
+            return Err(TapPlanError::Bind(TapBindError::DuplicateBind { name: tap.name.clone() }));
+        }
+    }
+
+    // Unknown-tap guard: every plan name must be declared.
+    for name in plan.by_name.keys() {
+        if !declared_taps.iter().any(|t| &t.name == name) {
+            return Err(TapPlanError::UnknownTap { name: name.clone() });
+        }
+    }
+
+    // Resolve each declared tap to its subscription; validate all Named
+    // subscriptions fully before any store I/O.
+    enum Resolved {
+        Named { label: String, params: Vec<(String, Scalar)> },
+        Live(Box<dyn FnMut(Timestamp, Cell) + Send>),
+    }
+    let mut resolved: Vec<(String, ScalarKind, Resolved)> = Vec::new();
+    for tap in &declared_taps {
+        let kind = flat.signatures[tap.node].output[tap.field].kind;
+        let sub = match plan.by_name.remove(&tap.name) {
+            Some(TapSubscription::Named { label, params }) => Resolved::Named { label, params },
+            Some(TapSubscription::Live(consumer)) => Resolved::Live(consumer),
+            None => match &plan.default_named {
+                Some((label, params)) => {
+                    Resolved::Named { label: label.clone(), params: params.clone() }
+                }
+                None => continue, // unbound, inert (C27)
+            },
+        };
+        if let Resolved::Named { label, params } = &sub {
+            let entry = plan.registry.get(label).ok_or_else(|| TapPlanError::UnknownLabel {
+                label: label.clone(),
+                roster: plan.registry.labels(),
+            })?;
+            if !(entry.binds_at)(kind) {
+                return Err(TapPlanError::KindMismatch {
+                    tap: tap.name.clone(),
+                    label: label.clone(),
+                    kind,
+                });
+            }
+            validate_params(entry, params)?;
+        }
+        resolved.push((tap.name.clone(), kind, sub));
+    }
+
+    // Anything persisting? Then begin the run (directory creation is the
+    // pre-run refusal point).
+    let persists = resolved.iter().any(|(_, _, s)| matches!(s, Resolved::Named { .. }));
+    let streamer = if persists { Some(env.trace_store().begin_run(run_name)?) } else { None };
+
+    // Bind one consumer per resolved tap.
+    let mut bound = BoundTaps {
+        streamer,
+        declared: declared_taps.iter().map(|t| t.name.clone()).collect(),
+        persisted: Vec::new(),
+        rows: Vec::new(),
+        outcomes: Vec::new(),
+    };
+    for (name, kind, sub) in resolved {
+        let node: Box<dyn Node> = match sub {
+            Resolved::Named { label, params } => {
+                let entry = plan.registry.get(&label).expect("validated above");
+                bound.persisted.push(name.clone());
+                match (entry.output)(kind) {
+                    FoldOutput::Series => {
+                        let opener = bound
+                            .streamer
+                            .as_ref()
+                            .expect("persisting tap implies a begun streamer")
+                            .register_tap(&name, kind);
+                        let (tx, rx) = mpsc::channel();
+                        bound.outcomes.push((name.clone(), rx));
+                        (entry.build)(FoldBuildCtx { kind, params, sink: FoldSink::Series(opener, tx) })
+                    }
+                    FoldOutput::Row(out_kind) => {
+                        let (tx, rx) = mpsc::channel();
+                        bound.rows.push((name.clone(), out_kind, rx));
+                        (entry.build)(FoldBuildCtx { kind, params, sink: FoldSink::Row(tx) })
+                    }
+                }
+            }
+            Resolved::Live(consumer) => Box::new(TapLive::new(kind, consumer)),
+        };
+        flat.bind_tap(&name, node, tap_sink_schema(kind))
+            .expect("declared tap binds (name from flat.taps, kind from its own signature)");
+    }
+    Ok(bound)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use aura_core::{AnyColumn, Ctx};
+    use std::sync::mpsc;
+
+    #[test]
+    fn roster_lists_the_eight_core_entries_with_docs() {
+        let r = FoldRegistry::core();
+        let labels: Vec<&str> = r.roster().iter().map(|(l, _)| *l).collect();
+        // BTreeMap order — alphabetical.
+        assert_eq!(
+            labels,
+            vec!["count", "first", "last", "max", "mean", "min", "record", "sum"]
+        );
+        assert!(r.roster().iter().all(|(_, doc)| !doc.is_empty()), "every entry documents itself");
+    }
+
+    #[test]
+    fn unknown_label_refusal_enumerates_the_roster() {
+        let r = FoldRegistry::core();
+        let e = TapPlanError::UnknownLabel { label: "p95".to_string(), roster: r.labels() };
+        let msg = e.to_string();
+        assert!(msg.starts_with("unknown fold 'p95'"), "{msg}");
+        assert!(msg.contains("record") && msg.contains("mean"), "roster enumerated: {msg}");
+    }
+
+    #[test]
+    fn arithmetic_folds_bind_only_at_f64_record_and_count_anywhere() {
+        let r = FoldRegistry::core();
+        for label in ["sum", "mean", "min", "max"] {
+            let e = r.get(label).expect(label);
+            assert!((e.binds_at)(ScalarKind::F64), "{label} binds f64");
+            assert!(!(e.binds_at)(ScalarKind::I64), "{label} refuses i64");
+            assert!(!(e.binds_at)(ScalarKind::Bool), "{label} refuses bool");
+        }
+        for label in ["record", "count", "first", "last"] {
+            let e = r.get(label).expect(label);
+            for k in [ScalarKind::F64, ScalarKind::I64, ScalarKind::Bool, ScalarKind::Timestamp] {
+                assert!((e.binds_at)(k), "{label} binds {k:?}");
+            }
+        }
+    }
+
+    #[test]
+    fn a_param_on_a_param_less_entry_is_refused() {
+        let r = FoldRegistry::core();
+        let e = validate_params(
+            r.get("mean").expect("mean"),
+            &[("q".to_string(), Scalar::f64(0.95))],
+        )
+        .expect_err("core entries take no params");
+        assert_eq!(e.to_string(), "fold 'mean' takes no parameters (got 'q')");
+    }
+
+    #[test]
+    fn a_layer_registered_entry_resolves_and_builds() {
+        // The dependency-injection proof without a trading layer: a test-only
+        // entry registered on top of core resolves by label and its consumer
+        // runs (here: a re-labelled count over the committed TapFold core).
+        let mut r = FoldRegistry::core();
+        r.register(FoldEntry {
+            label: "tally",
+            doc: "test-only: warm-row count under a layer label",
+            params: Vec::new(),
+            binds_at: Box::new(|_| true),
+            output: Box::new(|_| FoldOutput::Row(ScalarKind::I64)),
+            build: Box::new(|ctx| {
+                let FoldSink::Row(tx) = ctx.sink else {
+                    panic!("wiring contract: tally is a Row entry");
+                };
+                Box::new(TapFold::new(ctx.kind, FoldKind::Count, tx))
+            }),
+        });
+        assert_eq!(r.labels().len(), 9, "core plus the layer entry");
+        let entry = r.get("tally").expect("layer entry resolves");
+
+        let (tx, rx) = mpsc::channel();
+        let mut node = (entry.build)(FoldBuildCtx {
+            kind: ScalarKind::F64,
+            params: Vec::new(),
+            sink: FoldSink::Row(tx),
+        });
+        let mut inputs = vec![AnyColumn::with_capacity(ScalarKind::F64, 1)];
+        node.initialize();
+        for (t, v) in [(1_i64, 5.0_f64), (2, 7.0)] {
+            inputs[0].push(Scalar::f64(v)).unwrap();
+            assert_eq!(node.eval(Ctx::new(&inputs, Timestamp(t))), None);
+        }
+        node.finalize();
+        let rows: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
+        assert_eq!(rows, vec![(Timestamp(2), vec![Scalar::i64(2)])]);
+    }
+
+    /// Exercises the `record` entry's Series-sink build path (destructure
+    /// `FoldSink::Series` -> construct `TapRecorder`), left uncovered by
+    /// `a_layer_registered_entry_resolves_and_builds` above, which only
+    /// drives a Row-sink entry.
+    #[test]
+    fn record_entry_builds_a_series_sink_that_persists_the_stream() {
+        let root = std::env::temp_dir()
+            .join(format!("aura-runner-tap_plan-record-{}", std::process::id()));
+        let _ = std::fs::remove_dir_all(&root);
+        std::fs::create_dir_all(&root).expect("create temp store root");
+        let store = aura_registry::TraceStore::open(&root);
+        let streamer = store.begin_run("plan").expect("begin run");
+        let opener = streamer.register_tap("t", ScalarKind::F64);
+
+        let r = FoldRegistry::core();
+        let entry = r.get("record").expect("record");
+        let (outcome_tx, outcome_rx) = mpsc::channel();
+        let mut node = (entry.build)(FoldBuildCtx {
+            kind: ScalarKind::F64,
+            params: Vec::new(),
+            sink: FoldSink::Series(opener, outcome_tx),
+        });
+        let mut inputs = vec![AnyColumn::with_capacity(ScalarKind::F64, 1)];
+        node.initialize();
+        for (t, v) in [(1_i64, 5.0_f64), (2, 7.0)] {
+            inputs[0].push(Scalar::f64(v)).unwrap();
+            assert_eq!(node.eval(Ctx::new(&inputs, Timestamp(t))), None);
+        }
+        node.finalize();
+
+        let outcomes: Vec<Result<(), TraceStoreError>> = outcome_rx.try_iter().collect();
+        assert_eq!(outcomes.len(), 1, "exactly one terminal outcome");
+        assert!(outcomes[0].is_ok(), "record's Series build path writes cleanly: {outcomes:?}");
+    }
+
+    #[test]
+    fn schema_params_are_validated_missing_mismatched_and_unknown() {
+        // A schema-carrying test entry reaches the validation branches v1's
+        // param-less core entries cannot: missing, kind-mismatched, and
+        // unknown-against-a-non-empty-schema.
+        let entry = FoldEntry {
+            label: "windowed",
+            doc: "test-only: carries a param schema",
+            params: vec![ParamSpec { name: "length".to_string(), kind: ScalarKind::I64 }],
+            binds_at: Box::new(|_| true),
+            output: Box::new(|_| FoldOutput::Row(ScalarKind::F64)),
+            build: Box::new(|_| panic!("validation-only entry is never built")),
+        };
+        let missing = validate_params(&entry, &[]).expect_err("length is required");
+        assert_eq!(missing.to_string(), "fold 'windowed' requires parameter 'length' (I64)");
+        let mismatched = validate_params(&entry, &[("length".to_string(), Scalar::f64(2.0))])
+            .expect_err("kind mismatch");
+        assert_eq!(mismatched.to_string(), "fold 'windowed' parameter 'length' is I64, got F64");
+        let unknown = validate_params(&entry, &[("q".to_string(), Scalar::i64(1))])
+            .expect_err("unknown param");
+        assert_eq!(unknown.to_string(), "fold 'windowed' has no parameter 'q' — takes: length");
+    }
+}

crates/aura-runner/src/tap_recorder.rs

@@ -0,0 +1,196 @@
+//! `TapRecorder` — the `record` fold-registry entry's consumer (C8 sink) for
+//! one declared tap: holds the streaming trace writer **in-graph** and
+//! appends `(ctx.now(), cell)` per fired warm cycle — 16 B, `Copy`, zero heap
+//! per cycle (#77 part 2); no writer thread, no channel (#283 design
+//! revision). Lives in the assembly crate, not `aura-std`, because the
+//! writer type is `aura-registry`'s (C28 layering). Lifecycle: `initialize`
+//! opens the writer (deferred acquisition; an open error is stored and the
+//! node degrades to inert — the run completes), `eval` appends (an I/O error
+//! mid-run likewise stores + degrades), `finalize` finishes the writer and
+//! reports exactly one terminal outcome over the wiring's channel.
+
+use aura_core::{Cell, Ctx, Node, ScalarKind};
+use aura_registry::{TapTraceWriter, TapWriterOpener, TraceStoreError};
+use aura_std::newest_cell;
+use std::sync::mpsc::Sender;
+
+/// A single-column recording sink around an in-graph streaming writer.
+pub struct TapRecorder {
+    kind: ScalarKind,
+    opener: Option<TapWriterOpener>,
+    writer: Option<TapTraceWriter>,
+    error: Option<TraceStoreError>,
+    emitted: bool,
+    outcome_tx: Sender<Result<(), TraceStoreError>>,
+}
+
+impl TapRecorder {
+    pub fn new(
+        kind: ScalarKind,
+        opener: TapWriterOpener,
+        outcome_tx: Sender<Result<(), TraceStoreError>>,
+    ) -> Self {
+        Self { kind, opener: Some(opener), writer: None, error: None, emitted: false, outcome_tx }
+    }
+}
+
+impl Node for TapRecorder {
+    fn lookbacks(&self) -> Vec<usize> {
+        vec![1]
+    }
+
+    fn initialize(&mut self) {
+        match self.opener.take().expect("initialize runs once per run").open() {
+            Ok(w) => self.writer = Some(w),
+            Err(e) => self.error = Some(e),
+        }
+    }
+
+    fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> {
+        let cell = newest_cell(self.kind, &ctx)?; // None until warm
+        if let Some(w) = self.writer.as_mut()
+            && let Err(e) = w.append(ctx.now(), cell)
+        {
+            // I/O failure mid-run: store it, go inert, keep consuming —
+            // the run completes; the failure surfaces once at finalize.
+            self.error = Some(e);
+            self.writer = None;
+        }
+        None
+    }
+
+    fn finalize(&mut self) {
+        if self.emitted {
+            return;
+        }
+        self.emitted = true;
+        let outcome = match (self.writer.take(), self.error.take()) {
+            (Some(w), None) => w.finish(),
+            (_, Some(e)) => Err(e),
+            (None, None) => Ok(()), // finalize without initialize: benign
+        };
+        let _ = self.outcome_tx.send(outcome);
+    }
+
+    fn label(&self) -> String {
+        "TapRecorder".to_string()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use aura_core::{AnyColumn, Scalar, Timestamp};
+    use aura_registry::TraceStore;
+    use std::path::PathBuf;
+    use std::sync::mpsc;
+
+    /// A fresh unique store root under the OS temp dir (aura-runner has no
+    /// tempfile dev-dep; mirror of trace_store's own temp_traces_root).
+    fn temp_store_root(name: &str) -> PathBuf {
+        let dir = std::env::temp_dir()
+            .join(format!("aura-runner-taprec-{name}-{}", std::process::id()));
+        let _ = std::fs::remove_dir_all(&dir);
+        std::fs::create_dir_all(&dir).expect("create temp store root");
+        dir
+    }
+
+    #[test]
+    fn initialize_opens_records_after_warmup_and_reports_one_ok_outcome() {
+        let root = temp_store_root("happy");
+        let store = TraceStore::open(&root);
+        let streamer = store.begin_run("rec").expect("begin");
+        let (tx, rx) = mpsc::channel();
+        let mut sink =
+            TapRecorder::new(ScalarKind::F64, streamer.register_tap("t", ScalarKind::F64), tx);
+        let mut inputs = vec![AnyColumn::with_capacity(ScalarKind::F64, 1)];
+
+        sink.initialize();
+        // cold: nothing appended.
+        assert_eq!(sink.eval(Ctx::new(&inputs, Timestamp(1))), None);
+
+        for (t, v) in [(2_i64, 10.0_f64), (3, 20.0), (4, 30.0)] {
+            inputs[0].push(Scalar::f64(v)).unwrap();
+            assert_eq!(sink.eval(Ctx::new(&inputs, Timestamp(t))), None);
+        }
+        sink.finalize();
+        sink.finalize(); // second flush must not emit a second outcome
+
+        let outcomes: Vec<Result<(), TraceStoreError>> = rx.try_iter().collect();
+        assert_eq!(outcomes.len(), 1, "exactly one terminal outcome");
+        assert!(outcomes[0].is_ok(), "happy path outcome: {outcomes:?}");
+
+        let text = std::fs::read_to_string(root.join("traces").join("rec").join("t.json"))
+            .expect("finished tap file");
+        let v: serde_json::Value = serde_json::from_str(&text).expect("parse");
+        assert_eq!(v["tap"], "t");
+        assert_eq!(v["ts"], serde_json::json!([2, 3, 4]));
+        assert_eq!(v["columns"], serde_json::json!([[10.0, 20.0, 30.0]]));
+    }
+
+    #[test]
+    fn a_failed_open_degrades_to_inert_and_reports_one_err_outcome() {
+        let root = temp_store_root("failopen");
+        let store = TraceStore::open(&root);
+        let streamer = store.begin_run("rec").expect("begin");
+        let opener = streamer.register_tap("t", ScalarKind::F64);
+        // Make the deferred open fail: remove the run dir before initialize.
+        std::fs::remove_dir_all(root.join("traces").join("rec")).expect("remove run dir");
+
+        let (tx, rx) = mpsc::channel();
+        let mut sink = TapRecorder::new(ScalarKind::F64, opener, tx);
+        let mut inputs = vec![AnyColumn::with_capacity(ScalarKind::F64, 1)];
+
+        sink.initialize();
+        // inert but alive: eval consumes without panicking.
+        inputs[0].push(Scalar::f64(1.0)).unwrap();
+        assert_eq!(sink.eval(Ctx::new(&inputs, Timestamp(2))), None);
+        sink.finalize();
+
+        let outcomes: Vec<Result<(), TraceStoreError>> = rx.try_iter().collect();
+        assert_eq!(outcomes.len(), 1, "exactly one terminal outcome");
+        assert!(outcomes[0].is_err(), "the stored open error surfaces terminally");
+    }
+
+    /// Exercises the mid-run branch at `eval` (lines 51-57): a successfully
+    /// opened writer whose *write* later faults must store the error, drop
+    /// `self.writer` (go inert — no further append attempts), and still
+    /// report exactly one terminal `Err` at `finalize`, same as a failed
+    /// open. `/dev/full` always faults on write, so pointing the ts temp
+    /// stream at it (after a normal open) is a deterministic mid-run I/O
+    /// fault once the `BufWriter`'s internal buffer forces a real flush.
+    #[test]
+    #[cfg(unix)]
+    fn a_mid_run_append_failure_degrades_to_inert_and_reports_one_err_outcome() {
+        let root = temp_store_root("appendfail");
+        let store = TraceStore::open(&root);
+        let streamer = store.begin_run("rec").expect("begin");
+        let opener = streamer.register_tap("t", ScalarKind::F64);
+        // Retarget the ts temp stream at a permanently-full device: `open`
+        // still succeeds (create/truncate are no-ops on a char device), only
+        // a later flushed write faults.
+        std::os::unix::fs::symlink(
+            "/dev/full",
+            root.join("traces").join("rec").join("t.ts.tmp"),
+        )
+        .expect("symlink t.ts.tmp to /dev/full");
+
+        let (tx, rx) = mpsc::channel();
+        let mut sink = TapRecorder::new(ScalarKind::F64, opener, tx);
+        let mut inputs = vec![AnyColumn::with_capacity(ScalarKind::F64, 1)];
+
+        sink.initialize();
+        // Enough warm evals to force the BufWriter past its internal
+        // capacity and trigger a real (faulting) flush; `eval` never emits a
+        // row regardless of the underlying write outcome.
+        for t in 0..20_000_i64 {
+            inputs[0].push(Scalar::f64(1.0)).unwrap();
+            assert_eq!(sink.eval(Ctx::new(&inputs, Timestamp(t))), None);
+        }
+        sink.finalize();
+
+        let outcomes: Vec<Result<(), TraceStoreError>> = rx.try_iter().collect();
+        assert_eq!(outcomes.len(), 1, "exactly one terminal outcome");
+        assert!(outcomes[0].is_err(), "the mid-run write fault surfaces terminally");
+    }
+}

crates/aura-std/src/lib.rs

@@ -43,6 +43,9 @@ mod sign;
 mod sma;
 mod sqrt;
 mod sub;
+mod tap_cell;
+mod tap_fold;
+mod tap_live;
 mod when;
 pub use abs::Abs;
 pub use add::Add;
@@ -70,4 +73,7 @@ pub use sign::Sign;
 pub use sma::Sma;
 pub use sqrt::Sqrt;
 pub use sub::Sub;
+pub use tap_cell::newest_cell;
+pub use tap_fold::{fold_binds_at, fold_output_kind, FoldKind, TapFold};
+pub use tap_live::TapLive;
 pub use when::When;

crates/aura-std/src/tap_cell.rs

@@ -0,0 +1,18 @@
+//! Shared single-column read for the tap sinks (`TapFold`, `TapLive`, and
+//! the assembly crate's `aura_runner::TapRecorder`): the kind-dispatched
+//! newest-value read `Recorder::eval` does, re-encoded as a bare `Cell` (the
+//! C7 kind-erased carrier) instead of a kind-tagged `Scalar` — the
+//! #77-part-2 zero-heap payload.
+
+use aura_core::{Cell, Ctx, ScalarKind};
+
+/// The newest value of input column 0, as a `Cell`. `None` until the column
+/// is warm (the `?` on the window read), mirroring `Recorder`'s warm-up.
+pub fn newest_cell(kind: ScalarKind, ctx: &Ctx<'_>) -> Option<Cell> {
+    Some(match kind {
+        ScalarKind::F64 => Cell::from_f64(ctx.f64_in(0).get(0)?),
+        ScalarKind::I64 => Cell::from_i64(ctx.i64_in(0).get(0)?),
+        ScalarKind::Bool => Cell::from_bool(ctx.bool_in(0).get(0)?),
+        ScalarKind::Timestamp => Cell::from_ts(ctx.ts_in(0).get(0)?),
+    })
+}

crates/aura-std/src/tap_fold.rs

@@ -0,0 +1,263 @@
+//! `TapFold` — a folding sink (C8) over one declared-tap column: accumulates
+//! owned per-cycle state (no channel traffic, no heap per cycle) and emits
+//! exactly one summary row on `finalize` (the #138 lifecycle; `SeriesReducer`
+//! precedent). The fold is named from the closed [`FoldKind`] vocabulary
+//! (C25: closed, typed; a new aggregate is a new named entry, never inline
+//! logic in a data artifact). C7-pure: owned state + the channel handle.
+
+use aura_core::{Cell, Ctx, Node, Scalar, ScalarKind, Timestamp};
+use std::sync::mpsc::Sender;
+
+use crate::tap_cell::newest_cell;
+
+/// The closed fold vocabulary. Bind rules: `Sum`/`Mean`/`Min`/`Max` bind only
+/// at f64 taps ([`fold_binds_at`]); `Count` at any kind; `First`/`Last`
+/// kind-preserving at any kind. Output kinds: [`fold_output_kind`].
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum FoldKind {
+    Count,
+    Sum,
+    Mean,
+    Min,
+    Max,
+    First,
+    Last,
+}
+
+/// May `fold` bind at a tap of column kind `kind`? The arithmetic folds are
+/// f64-only; `Count`/`First`/`Last` are kind-agnostic.
+pub fn fold_binds_at(fold: FoldKind, kind: ScalarKind) -> bool {
+    match fold {
+        FoldKind::Sum | FoldKind::Mean | FoldKind::Min | FoldKind::Max => {
+            kind == ScalarKind::F64
+        }
+        FoldKind::Count | FoldKind::First | FoldKind::Last => true,
+    }
+}
+
+/// The output column kind of `fold` at a tap of column kind `kind`:
+/// `Count` → i64; the arithmetic folds → f64; `First`/`Last` preserve `kind`.
+pub fn fold_output_kind(fold: FoldKind, kind: ScalarKind) -> ScalarKind {
+    match fold {
+        FoldKind::Count => ScalarKind::I64,
+        FoldKind::Sum | FoldKind::Mean | FoldKind::Min | FoldKind::Max => ScalarKind::F64,
+        FoldKind::First | FoldKind::Last => kind,
+    }
+}
+
+/// Owned accumulator for every fold kind at once (a handful of words; keeping
+/// it total avoids a per-kind state enum). Sequential `sum += v` matches a
+/// slice-driven mean's operation order, so streamed and slice `Mean` agree
+/// bit-for-bit (the `SeriesFold` discipline, IEEE-754 non-associativity
+/// respected).
+#[derive(Clone, Copy, Debug)]
+struct FoldState {
+    count: u64,
+    sum: f64,
+    min: f64,
+    max: f64,
+    first: Option<(Timestamp, Cell)>,
+    last: Option<Cell>,
+}
+
+impl FoldState {
+    fn new() -> Self {
+        Self {
+            count: 0,
+            sum: 0.0,
+            min: f64::INFINITY,
+            max: f64::NEG_INFINITY,
+            first: None,
+            last: None,
+        }
+    }
+
+    fn fold(&mut self, ts: Timestamp, cell: Cell, kind: ScalarKind) {
+        self.count += 1;
+        if kind == ScalarKind::F64 {
+            let v = cell.f64();
+            self.sum += v;
+            if v < self.min {
+                self.min = v;
+            }
+            if v > self.max {
+                self.max = v;
+            }
+        }
+        if self.first.is_none() {
+            self.first = Some((ts, cell));
+        }
+        self.last = Some(cell);
+    }
+
+    fn folded_any(&self) -> bool {
+        self.count > 0
+    }
+
+    /// The summary value; call only when `folded_any()`.
+    fn output(&self, fold: FoldKind, kind: ScalarKind) -> Scalar {
+        match fold {
+            FoldKind::Count => Scalar::i64(self.count as i64),
+            FoldKind::Sum => Scalar::f64(self.sum),
+            FoldKind::Mean => Scalar::f64(self.sum / self.count as f64),
+            FoldKind::Min => Scalar::f64(self.min),
+            FoldKind::Max => Scalar::f64(self.max),
+            FoldKind::First => {
+                Scalar::from_cell(kind, self.first.expect("folded_any checked").1)
+            }
+            FoldKind::Last => Scalar::from_cell(kind, self.last.expect("folded_any checked")),
+        }
+    }
+}
+
+/// A single-column folding sink. Emits exactly one row on `finalize` (never
+/// twice, never on a never-warm tap): `(row_ts, [output])`, where `row_ts` is
+/// the last folded value's ts — except `First`, which carries the first warm
+/// value's ts (the instant its reported value was determined).
+pub struct TapFold {
+    kind: ScalarKind,
+    fold: FoldKind,
+    state: FoldState,
+    last_ts: Timestamp,
+    emitted: bool,
+    tx: Sender<(Timestamp, Vec<Scalar>)>,
+}
+
+impl TapFold {
+    pub fn new(kind: ScalarKind, fold: FoldKind, tx: Sender<(Timestamp, Vec<Scalar>)>) -> Self {
+        debug_assert!(fold_binds_at(fold, kind), "caller validates fold/kind before binding");
+        Self { kind, fold, state: FoldState::new(), last_ts: Timestamp(0), emitted: false, tx }
+    }
+}
+
+impl Node for TapFold {
+    fn lookbacks(&self) -> Vec<usize> {
+        vec![1]
+    }
+
+    fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> {
+        let cell = newest_cell(self.kind, &ctx)?; // None until warm
+        self.state.fold(ctx.now(), cell, self.kind);
+        self.last_ts = ctx.now();
+        None
+    }
+
+    fn finalize(&mut self) {
+        if self.state.folded_any() && !self.emitted {
+            self.emitted = true;
+            let ts = match self.fold {
+                FoldKind::First => self.state.first.expect("folded_any checked").0,
+                _ => self.last_ts,
+            };
+            let _ = self.tx.send((ts, vec![self.state.output(self.fold, self.kind)]));
+        }
+    }
+
+    fn label(&self) -> String {
+        format!("TapFold({:?})", self.fold)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use aura_core::AnyColumn;
+    use std::sync::mpsc;
+
+    /// Drive a fold over an f64 series and return the emitted rows.
+    fn run_f64_fold(fold: FoldKind, values: &[f64]) -> Vec<(Timestamp, Vec<Scalar>)> {
+        let (tx, rx) = mpsc::channel();
+        let mut sink = TapFold::new(ScalarKind::F64, fold, tx);
+        let mut inputs = vec![AnyColumn::with_capacity(ScalarKind::F64, 1)];
+        for (i, &v) in values.iter().enumerate() {
+            inputs[0].push(Scalar::f64(v)).unwrap();
+            assert_eq!(sink.eval(Ctx::new(&inputs, Timestamp(i as i64 + 1))), None);
+        }
+        sink.finalize();
+        rx.try_iter().collect()
+    }
+
+    #[test]
+    fn mean_matches_slice_mean_bit_for_bit() {
+        let values = [0.1, 0.2, 0.30000000000000004, -1.5, 2.75];
+        let rows = run_f64_fold(FoldKind::Mean, &values);
+        let slice_mean = values.iter().sum::<f64>() / values.len() as f64;
+        assert_eq!(rows, vec![(Timestamp(5), vec![Scalar::f64(slice_mean)])]);
+    }
+
+    #[test]
+    fn count_sum_min_max_over_a_known_series() {
+        let values = [3.0, -1.0, 2.0];
+        assert_eq!(
+            run_f64_fold(FoldKind::Count, &values),
+            vec![(Timestamp(3), vec![Scalar::i64(3)])]
+        );
+        assert_eq!(
+            run_f64_fold(FoldKind::Sum, &values),
+            vec![(Timestamp(3), vec![Scalar::f64(4.0)])]
+        );
+        assert_eq!(
+            run_f64_fold(FoldKind::Min, &values),
+            vec![(Timestamp(3), vec![Scalar::f64(-1.0)])]
+        );
+        assert_eq!(
+            run_f64_fold(FoldKind::Max, &values),
+            vec![(Timestamp(3), vec![Scalar::f64(3.0)])]
+        );
+    }
+
+    #[test]
+    fn first_carries_the_first_warm_ts_and_last_the_last() {
+        let values = [7.0, 8.0, 9.0];
+        assert_eq!(
+            run_f64_fold(FoldKind::First, &values),
+            vec![(Timestamp(1), vec![Scalar::f64(7.0)])],
+            "First reports the first warm value at ITS ts"
+        );
+        assert_eq!(
+            run_f64_fold(FoldKind::Last, &values),
+            vec![(Timestamp(3), vec![Scalar::f64(9.0)])]
+        );
+    }
+
+    #[test]
+    fn never_warm_tap_emits_nothing_and_finalize_emits_once() {
+        let (tx, rx) = mpsc::channel();
+        let mut sink = TapFold::new(ScalarKind::F64, FoldKind::Count, tx);
+        let inputs = vec![AnyColumn::with_capacity(ScalarKind::F64, 1)];
+        // cold eval, then finalize: nothing emitted.
+        assert_eq!(sink.eval(Ctx::new(&inputs, Timestamp(1))), None);
+        sink.finalize();
+        assert!(rx.try_recv().is_err(), "a never-warm fold emits no row");
+
+        // warm one value; a double finalize still emits exactly one row.
+        let (tx2, rx2) = mpsc::channel();
+        let mut sink2 = TapFold::new(ScalarKind::I64, FoldKind::Last, tx2);
+        let mut inputs2 = vec![AnyColumn::with_capacity(ScalarKind::I64, 1)];
+        inputs2[0].push(Scalar::i64(42)).unwrap();
+        assert_eq!(sink2.eval(Ctx::new(&inputs2, Timestamp(9))), None);
+        sink2.finalize();
+        sink2.finalize();
+        let rows: Vec<_> = rx2.try_iter().collect();
+        assert_eq!(rows, vec![(Timestamp(9), vec![Scalar::i64(42)])]);
+    }
+
+    #[test]
+    fn bind_rules_and_output_kinds_are_the_specced_table() {
+        for fold in [FoldKind::Sum, FoldKind::Mean, FoldKind::Min, FoldKind::Max] {
+            assert!(fold_binds_at(fold, ScalarKind::F64));
+            assert!(!fold_binds_at(fold, ScalarKind::I64));
+            assert!(!fold_binds_at(fold, ScalarKind::Bool));
+            assert!(!fold_binds_at(fold, ScalarKind::Timestamp));
+            assert_eq!(fold_output_kind(fold, ScalarKind::F64), ScalarKind::F64);
+        }
+        for kind in [ScalarKind::F64, ScalarKind::I64, ScalarKind::Bool, ScalarKind::Timestamp] {
+            assert!(fold_binds_at(FoldKind::Count, kind));
+            assert!(fold_binds_at(FoldKind::First, kind));
+            assert!(fold_binds_at(FoldKind::Last, kind));
+            assert_eq!(fold_output_kind(FoldKind::Count, kind), ScalarKind::I64);
+            assert_eq!(fold_output_kind(FoldKind::First, kind), kind);
+            assert_eq!(fold_output_kind(FoldKind::Last, kind), kind);
+        }
+    }
+}

crates/aura-std/src/tap_live.rs

@@ -0,0 +1,88 @@
+//! `TapLive` — the live-subscription sink (C8) for one declared tap: calls a
+//! consumer-owned closure `(Timestamp, Cell)` inline per fired warm cycle, on
+//! the sim thread, zero alloc. Loss policy is the closure's own (a UI
+//! consumer typically `try_send`s into its own channel and drops when behind)
+//! — per #283: consumer-owned logic and state, no registry ceremony.
+
+use aura_core::{Cell, Ctx, Node, ScalarKind, Timestamp};
+
+use crate::tap_cell::newest_cell;
+
+/// A single-column live sink around a consumer closure.
+pub struct TapLive {
+    kind: ScalarKind,
+    consumer: Box<dyn FnMut(Timestamp, Cell) + Send>,
+}
+
+impl TapLive {
+    pub fn new(kind: ScalarKind, consumer: impl FnMut(Timestamp, Cell) + Send + 'static) -> Self {
+        Self { kind, consumer: Box::new(consumer) }
+    }
+}
+
+impl Node for TapLive {
+    fn lookbacks(&self) -> Vec<usize> {
+        vec![1]
+    }
+
+    fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> {
+        let cell = newest_cell(self.kind, &ctx)?; // None until warm
+        (self.consumer)(ctx.now(), cell);
+        None
+    }
+
+    fn label(&self) -> String {
+        "TapLive".to_string()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use aura_core::{AnyColumn, Scalar};
+    use std::sync::mpsc;
+
+    #[test]
+    fn calls_the_consumer_per_warm_cycle_in_order_and_not_before() {
+        let (tx, rx) = mpsc::channel();
+        let mut sink = TapLive::new(ScalarKind::F64, move |ts, cell| {
+            let _ = tx.send((ts, cell.f64()));
+        });
+        let mut inputs = vec![AnyColumn::with_capacity(ScalarKind::F64, 1)];
+
+        // cold: consumer not called.
+        assert_eq!(sink.eval(Ctx::new(&inputs, Timestamp(1))), None);
+        assert!(rx.try_recv().is_err(), "no call before warm-up");
+
+        for (t, v) in [(2_i64, 1.5_f64), (3, 2.5)] {
+            inputs[0].push(Scalar::f64(v)).unwrap();
+            assert_eq!(sink.eval(Ctx::new(&inputs, Timestamp(t))), None);
+        }
+        let got: Vec<(Timestamp, f64)> = rx.try_iter().collect();
+        assert_eq!(got, vec![(Timestamp(2), 1.5), (Timestamp(3), 2.5)]);
+    }
+
+    /// `newest_cell`'s four-kind dispatch (shared with `TapFold`) is
+    /// otherwise only exercised at F64/I64; the Bool and Timestamp arms
+    /// need a witness too, so a regression in either goes caught.
+    #[test]
+    fn calls_the_consumer_at_bool_and_timestamp_kinds_too() {
+        let (tx, rx) = mpsc::channel();
+        let mut sink = TapLive::new(ScalarKind::Bool, move |ts, cell| {
+            let _ = tx.send((ts, cell.bool()));
+        });
+        let mut inputs = vec![AnyColumn::with_capacity(ScalarKind::Bool, 1)];
+        inputs[0].push(Scalar::bool(true)).unwrap();
+        assert_eq!(sink.eval(Ctx::new(&inputs, Timestamp(1))), None);
+        assert_eq!(rx.try_recv(), Ok((Timestamp(1), true)));
+
+        let (tx2, rx2) = mpsc::channel();
+        let mut sink2 = TapLive::new(ScalarKind::Timestamp, move |ts, cell| {
+            let _ = tx2.send((ts, cell.ts()));
+        });
+        let mut inputs2 = vec![AnyColumn::with_capacity(ScalarKind::Timestamp, 1)];
+        inputs2[0].push(Scalar::ts(Timestamp(99))).unwrap();
+        assert_eq!(sink2.eval(Ctx::new(&inputs2, Timestamp(1))), None);
+        assert_eq!(rx2.try_recv(), Ok((Timestamp(1), Timestamp(99))));
+    }
+}

docs/design/contracts/c08-node-contract.history.md

@@ -157,3 +157,17 @@ text — `\n`/`\t`/`\r` named, other control chars as `\u00XX`), the viewer show
 it in both composite view states (collapsed tooltip, expanded cluster frame) and
 the root's as a muted header line. The construction op-script vocabulary
 deliberately has no doc-carrying surface yet (scope cut recorded on #125).
+
+**Refinement (2026-07-21 — start-of-stream `initialize` lifecycle; #77 resolved,
+#283).** The contract gains `finalize`'s mirror: `initialize()`, default no-op,
+called once per node in topological order before the first source value (a
+`Harness::run` prologue with its own once-per-run mirror test). Consumers acquire
+run resources there — the in-graph record consumer (`TapRecorder`, `aura-runner`)
+opens its streaming trace writer in `initialize`, appends per fired warm cycle,
+and finishes/reports exactly one terminal outcome at `finalize`; the hook is
+infallible by signature, so an acquisition failure degrades the node to inert and
+surfaces once, terminally, at `finalize`. This closes the #77 line the cycle-0070
+entry left open: declared-tap consumers now move `(Timestamp, Cell)` (zero
+per-cycle heap), and the `Recorder`→`Probe` rename is retired — `Recorder` keeps
+its name on the legacy live/`--trace`/test-tap surface, whose migration onto the
+subscriber shape is tracked as #308.

docs/design/contracts/c08-node-contract.md

@@ -32,8 +32,10 @@ a zero-width `Some(&[])`, and it pushes its record to a destination it holds as
 field (a channel, a chart handle). A node may record **and** return a forwarded
 output in the same `eval` (the "both" case). `None` = filter / not-yet-warmed-up /
 pure sink. Sources are pure producers; sinks are pure consumers. Beside `eval`,
-the contract has a second lifecycle phase, `finalize()` — a default-no-op hook the
-engine calls once per node, in topological order, after the source loop drains —
+the contract has a symmetric lifecycle pair — default-no-op hooks the engine calls
+once per node, in topological order: `initialize()` before the first source value,
+letting a consumer acquire its run resources (e.g. the file a recording sink
+streams into) at stream start, and `finalize()` after the source loop drains,
 letting a sink fold online and flush one compact summary at stream end instead of
 streaming a row every fired cycle. A node also exposes `label()`, a single-line,
 non-load-bearing render symbol (C23).
@@ -55,7 +57,12 @@ boundary is the determinism / graph-as-data boundary (C1/C7). `finalize` gives a
 folding sink O(trades)/O(1) owned accumulator state and one summary row, instead of
 an unbounded channel that buffers O(cycles) rows until the run ends; it runs once
 *after* the deterministic event loop, adding no within-sim concurrency (C1), and
-holds only owned state, no interior mutability (C7).
+holds only owned state, no interior mutability (C7). `initialize` mirrors it at
+stream start (#283): resource acquisition happens inside the same deterministic
+sequence, and the hook is infallible by signature — a consumer that fails to
+acquire stores the error, degrades to inert, and surfaces the failure once,
+terminally, at `finalize` (the run completes; refusal-worthy faults are caught
+pre-run, at construction/bind time).
 
 ## Current state
 
@@ -70,7 +77,8 @@ structural axis, C20, never a numeric sweep param).
 `trait Node` declares `lookbacks() -> Vec<usize>`, `eval(&mut self, ctx) ->
 Option<&[Cell]>`, `label() -> String` (default a placeholder every shipped node
 overrides; overrides carry identifying params so `SMA(2)` vs `SMA(4)` disambiguate
-in a graph render, #13), and `finalize(&mut self)` (default no-op). There is **no**
+in a graph render, #13), and the lifecycle pair `initialize(&mut self)` /
+`finalize(&mut self)` (both default no-op). There is **no**
 `Node::schema()` — the signature is pre-build data, not a built-node method. The
 signature is declared once on the value-empty recipe `PrimitiveBuilder`
 (`aura-core/src/node.rs`), whose `schema()` / `params()` are read pre-build by the
@@ -83,9 +91,10 @@ against the producer's `output` at bootstrap. A sink's empty `output` therefore
 makes it structurally unwireable as an in-graph producer — no edge can bind a field
 of a zero-output node. The run loop (`aura-engine/src/harness.rs`) debug-asserts
 `row.len()` equals the declared output width, and that `lookbacks()` arity equals
-`signature().inputs.len()`. The end-of-stream `finalize` epilogue is realized in
-`Harness::run` (`aura-engine/src/harness.rs`), which calls `finalize()` once per
-node in topological order after the source loop drains; `GatedRecorder` and
+`signature().inputs.len()`. The lifecycle pair is realized in `Harness::run`
+(`aura-engine/src/harness.rs`): an `initialize()` prologue once per node in
+topological order before the source loop, and the `finalize()` epilogue after it
+drains — each pinned by its own once-per-run mirror test; `GatedRecorder` and
 `SeriesReducer` (`aura-std`) are the folding siblings of the per-cycle `Recorder`
 (`aura-std/src/recorder.rs`), which survives for the live / `--trace` / test-tap
 path.
@@ -124,8 +133,7 @@ untouched), blanked in the identity projection by `strip_debug_symbols`, dissolv
 at inline like the name, and surfaced read-only in the graph model and viewer
 (C22).
 
-Deferred: the recording sink's own per-cycle allocation — the `Recorder`→`Probe`
-rename and its accumulate-vs-stream choice — stays open (#77). Naming input ports
+Deferred: Naming input ports
 (`PortSpec.name`) makes swap-prone same-kind slots self-documenting but does not add
 a name-consuming wiring validation; a swapped same-kind wiring is still only
 kind-checked (#21). The construction op-script vocabulary has no doc-carrying

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

@@ -11,14 +11,22 @@ recursion), landing in `FlatGraph.taps` as a `FlatTap { name, node, field }` who
 name survives compile and is load-bearing for by-name binding (like
 `SourceSpec.role`, #275).
 
-Binding is **run-mode-aware**: the run-mode-owning layer constructs a sink (a
-`Recorder`) at a bound tap via `FlatGraph::bind_tap`, which takes a **caller-built**
+Binding is **run-mode-aware**: the run-mode-owning layer constructs a consumer at
+a bound tap via `FlatGraph::bind_tap`, which takes a **caller-built**
 `Box<dyn Node>` sink (so the engine keeps its `aura-core`-only production
 dependency — it never constructs a domain sink type) and appends it plus an edge
-before bootstrap. The single-run path (`run_signal_r`) binds and records each
-declared tap, persisting the series through the existing trace store
-(`env.trace_store()`), so the tap columns surface through the same tooling the
-campaign path feeds; a sweep/reduce run leaves taps unbound.
+before bootstrap. What consumes a tap is declared per run by a **tap plan**
+(#283): a `Named { label, params }` subscription resolves against a layered
+**fold registry** (core vocabulary `record | count | sum | mean | min | max |
+first | last`; growth is a new Rust entry, C25, injectable by higher layers
+without a core edit), and `Live(closure)` is the single deliberately non-data
+variant. Both declared-tap entry points — the single-run path (`run_signal_r`)
+and the bare measurement path (`run_measurement`) — bind through one shared
+wiring pair, so they cannot drift: `record` persists the full series at constant
+memory through the trace store's streamed write path (`env.trace_store()`), folds
+keep an O(1) accumulator and land one summary row at finalize, so the tap columns
+surface through the same tooling the campaign path feeds; a sweep/reduce run
+leaves taps unbound.
 
 **Forbids.** A tap carrying a channel endpoint or effect in the serialized artefact
 — recording policy is run-mode authority, not fragment-embedded (a fragment must
@@ -54,8 +62,19 @@ caller-built `Box<dyn Node>` sink plus an edge and raises a typed `UndeclaredTap
 on a tap the graph does not declare (duplicate detection across binds is the
 caller's — the method keeps no cross-call state). Lowering resolves and hoists taps
 via `resolve_tap_wire` and the `flat_taps` accumulator (`blueprint.rs`). The
-single-run bind/record path is `run_signal_r` in `aura-runner::member`; the
-sweep/reduce path never calls `bind_tap`.
+subscription seam is `aura-runner::tap_plan` (#283): `TapPlan`/`TapSubscription`,
+the layered `FoldRegistry` (each entry carries a `doc` line — the help surface
+and the roster-enumerating refusal — plus a scalar-typed param schema; all core
+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`
+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
+exactly one terminal outcome; fold consumers (`aura-std::TapFold`) land one
+summary row; live closures run inline (`aura-std::TapLive`). The sweep/reduce
+path never calls `bind_tap`.
 
 The chain-pruning benefit — a sweep paying zero for the study wires behind an
 unbound tap — is **deferred to the future DCE cycle** ([C23](c23-graph-compilation.md));

docs/design/contracts/c28-stratification.md

@@ -107,7 +107,14 @@ The workspace realizes the full ladder today, cut by layer, with only the
 
 The closed node roster lives in `aura-vocabulary`. The process column is
 `aura-registry` (C18) + `aura-research` + `aura-campaign`; the assembly position
-is `aura-runner`; the shell is `aura-cli`.
+is `aura-runner` — which since #283 also *defines* one graph node of its own, the
+in-graph record consumer `TapRecorder` (`aura-runner/src/tap_recorder.rs`): it
+holds an `aura-registry` streaming writer, so it can live neither in `aura-std`
+(engine layer, must not know the process column) nor in `aura-registry`
+(persistence, not graph vocabulary) — the assembly crate is exactly where graph
+and persistence meet, alongside the `tap_plan` wiring
+(`TapPlan`/`FoldRegistry`/`bind_tap_plan`) that binds it. The shell is
+`aura-cli`.
 
 The dependency direction obeys the rule across the whole stack. The engine names
 no concrete metric type: `aura_engine::RunReport<M>` is generic over its metric
@@ -160,10 +167,12 @@ demand-gated, no tracking issue): measurement runs as sweep-family citizens
 (report unification, campaign engine generic-over-`M`), until a concrete
 family/campaign demand exists.
 
-**Deferred.** ~20 refusal sites inside `aura-runner`'s single-run verb paths still
-terminate the process (`std::process::exit`); their conversion to `RunnerError`
-propagation is tracked as **#297** (the campaign path already refuses via
-`MemberFault`, never a process exit).
+**Deferred.** ~24 refusal sites inside `aura-runner`'s single-run verb paths still
+terminate the process (`std::process::exit`; the #283 tap-plan refusals added four
+— typed as `TapPlanError` before the exit, so the eventual conversion is a
+mechanical rewrap); their conversion to `RunnerError` propagation is tracked as
+**#297** (the campaign path already refuses via `MemberFault`, never a process
+exit).
 
 ## See also
 - [C1](c01-determinism.md) — determinism / bit-identity, the correctness invariant the layer cuts preserve

docs/glossary.md

@@ -318,11 +318,13 @@ The harness's structural parameterization — which strategy, instrument(s), bro
 An orchestration axis varying tuning params (grid or random) within a fixed structure. The inner, param-tuning loop, distinct from the structural experiment matrix. On a loaded blueprint every open knob (`--list-axes`) is **required** — a subset is refused with the missing knob named; pin an unwanted knob with a single-value axis (`--axis name=<one-value>`), there is no default. `aura sweep --axis` takes the `--list-axes`-printed, root-composite-wrapped name (e.g. `graph.fast.length`) — not the raw `param_space` name (`fast.length`) that `graph introspect --params` and a campaign document's axes use; the CLI strips exactly one leading wrapper segment (#210). The `aura sweep` CLI verb is now thin sugar over the `campaign document` path — its blueprint form (`<bp.json> --real`) translates to a generated, content-addressed campaign run through the one executor (#210); the built-in `--strategy` sweep surface was retired by #159 (its hard-wired harnesses removed) — sweep now runs from a blueprint + `--axis`, and a retired `--strategy` token falls to the generic usage error. A ganged pair contributes ONE axis.
 
 ### tap
-**Avoid:** —
+**Avoid:** probe, monitor, scope (for the observation slot — "probe" is taken by the sweep-terminal `blueprint_axis_probe` sense; the #77 `Recorder`→`Probe` rename was retired, 2026-07-21)
 A named recorded stream produced by a recording `sink` — the addressable label (e.g. `equity`, `net_r_equity`) under which one sink's per-cycle output is persisted as a columnar (SoA) `ColumnarTrace` and selected for charting via `--tap`. Distinct from the `sink` node that emits it (a tap is the stream, the sink is the role) and from a whole recorded run (a bundle of taps); taps fire at their own cadences and are fused only by joining on the recorded timestamp, never by positional index. In a `campaign document`, `persist_taps` names taps from the CLOSED vocabulary `equity | exposure | r_equity | net_r_equity` (`tap_vocabulary`, 0109/#201) — a new observable is a new vocabulary entry or an authored blueprint sink, never an open node-path namespace. Persisted taps are charted by the printed handle: a campaign run's via the record's `trace_name`, a `sweep`/`walkforward --trace` family's via the family handle the run prints (`aura chart <handle>`; its members are keyed `<cell>/<member>` in the chart) — or, equivalently, by the `--trace <NAME>` the user chose, when `NAME` uniquely resolves against the recorded campaign documents (#238; a name reused across runs refuses rather than guessing).
 
 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.
+
 ### topology hash
 **Avoid:** —
 The `content id` of a run's signal blueprint in its run-record role: stamped into the manifest as `topology_hash`, keying the reproduction store and anchoring `aura reproduce` — the same SHA-256 over the same canonical bytes. Distinct from the debug-name-blind `identity id`.

fieldtests/cycle-283-tap-subscribers/tap_1_record_spread.ops.json

@@ -0,0 +1,13 @@
+[
+  {"op": "source", "role": "price", "kind": "F64"},
+  {"op": "add", "type": "SMA", "name": "fast", "bind": {"length": {"I64": 2}}},
+  {"op": "add", "type": "SMA", "name": "slow", "bind": {"length": {"I64": 4}}},
+  {"op": "feed", "role": "price", "into": ["fast.series", "slow.series"]},
+  {"op": "add", "type": "Sub", "name": "sub"},
+  {"op": "connect", "from": "fast.value", "to": "sub.lhs"},
+  {"op": "connect", "from": "slow.value", "to": "sub.rhs"},
+  {"op": "add", "type": "Bias", "name": "bias", "bind": {"scale": {"F64": 0.5}}},
+  {"op": "connect", "from": "sub.value", "to": "bias.signal"},
+  {"op": "tap", "from": "sub.value", "as": "spread"},
+  {"op": "expose", "from": "bias.bias", "as": "bias"}
+]

fieldtests/cycle-283-tap-subscribers/tap_4_measure_ic.ops.json

@@ -0,0 +1,15 @@
+[
+  {"op": "source", "role": "price", "kind": "F64"},
+  {"op": "add", "type": "SMA", "name": "fast", "bind": {"length": {"I64": 2}}},
+  {"op": "add", "type": "SMA", "name": "slow", "bind": {"length": {"I64": 4}}},
+  {"op": "add", "type": "SMA", "name": "px", "bind": {"length": {"I64": 1}}},
+  {"op": "feed", "role": "price", "into": ["fast.series", "slow.series", "px.series"]},
+  {"op": "add", "type": "Sub", "name": "sub"},
+  {"op": "connect", "from": "fast.value", "to": "sub.lhs"},
+  {"op": "connect", "from": "slow.value", "to": "sub.rhs"},
+  {"op": "add", "type": "Bias", "name": "bias", "bind": {"scale": {"F64": 0.5}}},
+  {"op": "connect", "from": "sub.value", "to": "bias.signal"},
+  {"op": "tap", "from": "sub.value", "as": "signal"},
+  {"op": "tap", "from": "px.value", "as": "price"},
+  {"op": "expose", "from": "bias.bias", "as": "bias"}
+]

fieldtests/cycle-283-tap-subscribers/tap_consumer/.gitignore

@@ -0,0 +1,2 @@
+target/
+Cargo.lock

fieldtests/cycle-283-tap-subscribers/tap_consumer/Cargo.toml

@@ -0,0 +1,24 @@
+# A downstream World-program consumer of aura-runner's public tap-plan API.
+# Detached from the aura workspace via an empty [workspace] table, exactly as a
+# real research project's node crate is (docs/project-layout.md). It reaches the
+# fold and live tap subscriptions — the two surfaces neither CLI verb exposes
+# (both pass a record-all plan) — through run_signal_r + TapPlan.
+[package]
+name = "tap-consumer"
+version = "0.1.0"
+edition = "2021"
+
+[workspace]
+
+[dependencies]
+aura-core = { path = "../../../crates/aura-core" }
+aura-engine = { path = "../../../crates/aura-engine" }
+aura-runner = { path = "../../../crates/aura-runner" }
+
+[[bin]]
+name = "tap_2_fold"
+path = "src/tap_2_fold.rs"
+
+[[bin]]
+name = "tap_3_live"
+path = "src/tap_3_live.rs"

fieldtests/cycle-283-tap-subscribers/tap_consumer/examples/refusal_probe.rs

@@ -0,0 +1,21 @@
+//! Ergonomics probe for the fold axis: what does a downstream *library*
+//! consumer get when it subscribes an unknown fold label? A `Result` it can
+//! handle, or a whole-process termination? (C28 tracks the ~24 process-exit
+//! refusal sites as deferred #297.)
+//!
+//! Usage: cargo run --example refusal_probe -- <blueprint.json>
+
+use aura_engine::blueprint_from_json;
+use aura_runner::member::{run_signal_r, RunData};
+use aura_runner::project::Env;
+use aura_runner::{TapPlan, TapSubscription};
+
+fn main() {
+    let json = std::fs::read_to_string(std::env::args().nth(1).unwrap()).unwrap();
+    let env = Env::std();
+    let signal = blueprint_from_json(&json, &|id| env.resolve(id)).unwrap();
+    let mut plan = TapPlan::empty();
+    plan.subscribe("signal", TapSubscription::named("median")); // not in the core roster
+    let _ = run_signal_r(signal, &[], RunData::Synthetic, 0, &env, plan);
+    println!("RETURNED NORMALLY (no refusal surfaced to the caller)");
+}

fieldtests/cycle-283-tap-subscribers/tap_consumer/src/tap_2_fold.rs

@@ -0,0 +1,39 @@
+//! Axis 2 — fold subscription. A downstream consumer subscribes named core
+//! folds (`mean`, `count`) against the declared taps of a blueprint and runs it
+//! through the R scaffolding. Neither CLI verb can do this (both pass a
+//! record-all plan), so this is the only reachable path to a fold.
+//!
+//! Usage: tap_2_fold <blueprint.json>
+//! Run from a scratch dir; the fold rows land in ./runs/traces/<name>/.
+
+use aura_engine::blueprint_from_json;
+use aura_runner::member::{run_signal_r, RunData};
+use aura_runner::project::Env;
+use aura_runner::{TapPlan, TapSubscription};
+
+fn main() {
+    let path = std::env::args().nth(1).expect("usage: tap_2_fold <blueprint.json>");
+    let json = std::fs::read_to_string(&path).expect("read blueprint");
+
+    // Env::std() is the no-project context: every accessor collapses to the
+    // literal defaults (trace store rooted at ./runs). Its resolver doubles as
+    // the vocabulary resolver blueprint_from_json needs.
+    let env = Env::std();
+    let signal =
+        blueprint_from_json(&json, &|id| env.resolve(id)).expect("parse blueprint");
+
+    // Build a per-tap plan: mean of the signal, count of the price. empty() =
+    // no default, so any tap left unsubscribed stays inert (C27).
+    let mut plan = TapPlan::empty();
+    plan.subscribe("signal", TapSubscription::named("mean"));
+    plan.subscribe("price", TapSubscription::named("count"));
+
+    let report = run_signal_r(signal, &[], RunData::Synthetic, 0, &env, plan);
+
+    // The fold rows are persisted through the trace store, not returned in the
+    // report — so a consumer reads them back from ./runs/traces/<name>/.
+    println!("fold run complete (signal=mean, price=count)");
+    if let Some(r) = report.metrics.r {
+        println!("report expectancy_r = {}", r.expectancy_r);
+    }
+}

fieldtests/cycle-283-tap-subscribers/tap_consumer/src/tap_3_live.rs

@@ -0,0 +1,56 @@
+//! Axis 3 — live observation. A downstream consumer attaches an inline closure
+//! to a declared tap and observes every (Timestamp, Cell) on the sim thread,
+//! accumulating into shared state it reads after the run. No disk, no CLI path.
+//!
+//! Usage: tap_3_live <blueprint.json>
+
+use std::sync::{Arc, Mutex};
+
+use aura_engine::blueprint_from_json;
+use aura_runner::member::{run_signal_r, RunData};
+use aura_runner::project::Env;
+use aura_runner::{TapPlan, TapSubscription};
+
+#[derive(Default)]
+struct Seen {
+    n: u64,
+    sum: f64,
+    first_ts: Option<i64>,
+    last_ts: i64,
+}
+
+fn main() {
+    let path = std::env::args().nth(1).expect("usage: tap_3_live <blueprint.json>");
+    let json = std::fs::read_to_string(&path).expect("read blueprint");
+
+    let env = Env::std();
+    let signal =
+        blueprint_from_json(&json, &|id| env.resolve(id)).expect("parse blueprint");
+
+    let seen = Arc::new(Mutex::new(Seen::default()));
+    let sink = Arc::clone(&seen);
+
+    let mut plan = TapPlan::empty();
+    plan.subscribe(
+        "signal",
+        TapSubscription::live(move |ts, cell| {
+            let mut s = sink.lock().unwrap();
+            s.n += 1;
+            s.sum += cell.f64();
+            s.first_ts.get_or_insert(ts.0);
+            s.last_ts = ts.0;
+        }),
+    );
+
+    let _report = run_signal_r(signal, &[], RunData::Synthetic, 0, &env, plan);
+
+    let s = seen.lock().unwrap();
+    let mean = if s.n > 0 { s.sum / s.n as f64 } else { 0.0 };
+    println!(
+        "live saw {} cycles, ts {}..={}, mean(signal)={}",
+        s.n,
+        s.first_ts.unwrap_or(-1),
+        s.last_ts,
+        mean
+    );
+}