diff --git a/crates/aura-cli/src/main.rs b/crates/aura-cli/src/main.rs index 1fead8e..e24a859 100644 --- a/crates/aura-cli/src/main.rs +++ b/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, ¶ms, data, a.seed.unwrap_or(0), env); + let report = run_signal_r(signal, ¶ms, 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 @@ -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"); diff --git a/crates/aura-runner/src/lib.rs b/crates/aura-runner/src/lib.rs index 8f251ca..8621760 100644 --- a/crates/aura-runner/src/lib.rs +++ b/crates/aura-runner/src/lib.rs @@ -18,11 +18,16 @@ 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 diff --git a/crates/aura-runner/src/member.rs b/crates/aura-runner/src/member.rs index 02ab45b..9a60f36 100644 --- a/crates/aura-runner/src/member.rs +++ b/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 = BTreeSet::new(); - let mut tap_drains: Vec<(String, ScalarKind, mpsc::Receiver<(Timestamp, Vec)>)> = Vec::new(); - let declared: Vec = flat.taps.clone(); - for tap in &declared { - if !seen.insert(tap.name.clone()) { - eprintln!("aura: {}", aura_engine::TapBindError::DuplicateBind { name: tap.name.clone() }); - std::process::exit(1); - } - let kind = flat.signatures[tap.node].output[tap.field].kind; - let (tx, rx) = mpsc::channel(); - let sig = Recorder::builder(vec![kind], Firing::Any, tx.clone()).schema().clone(); - flat.bind_tap(&tap.name, Box::new(Recorder::new(&[kind], Firing::Any, tx)), sig) - .expect("declared tap binds (name from flat.taps, kind from its own signature)"); - tap_drains.push((tap.name.clone(), kind, rx)); - } + // 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 = tap_drains - .into_iter() - .map(|(name, kind, rx)| { - let rows: Vec<(Timestamp, Vec)> = rx.try_iter().collect(); - ColumnarTrace::from_rows(&name, &[kind], &rows) - }) - .collect(); - env.trace_store().write(&run_name, &manifest, &tap_traces).unwrap_or_else(|e| { - eprintln!("aura: writing tap traces failed: {e}"); - std::process::exit(1); - }); - } + // Close the tap plan: 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 } } @@ -1158,8 +1139,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 +1292,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"); } diff --git a/crates/aura-runner/src/tap_plan.rs b/crates/aura-runner/src/tap_plan.rs new file mode 100644 index 0000000..74541ce --- /dev/null +++ b/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), +} + +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, +} + +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)>), + Series(TapWriterOpener, Sender>), +} + +/// 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, + /// May this entry bind a tap column of this kind? + pub binds_at: Box bool>, + /// What the consumer persists, given the tap's column kind. + pub output: Box 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 Box>, +} + +/// 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 }, + /// 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 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 = 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, + declared: Vec, + /// Persisting taps (Series and Row alike), declared-tap order — the + /// `index.json` tap order. + persisted: Vec, + rows: Vec<(String, ScalarKind, Receiver<(Timestamp, Vec)>)>, + outcomes: Vec<(String, Receiver>)>, +} + +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)> = 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 { + let declared_taps: Vec = flat.taps.clone(); + + // Dedup guard (caller-owned per TapBindError::DuplicateBind's doc). + let mut seen: BTreeSet = 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), + } + 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 = 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)> = 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> = 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"); + } +}