claude
9221bcd167
A single run persisted its taps and then said nothing about where: the only way to learn the directory name was `ls runs/traces/`, and the chart intake's not-found refusal pointed at "the handle a sweep/walk-forward/campaign run printed" — two verbs retired with #319, and a single run printed no handle at all. A caller holding a family id from a families listing was stuck: that id is not a trace handle, and nothing said so. The handle now rides BESIDE the report, not inside it. Both declared-tap entry points return `RunOutcome { report, skipped, trace_name }` in place of the `(report, skipped)` pair, and the CLI composes report + handle into one stdout object through a `serde(flatten)` wrapper. `RunReport`, `MeasurementReport`, `RunManifest` and the registry's compat mirror are untouched, so no stored record shape moves and a tap-free run's line stays byte-identical. Placement is the load-bearing decision, and it follows a precedent rather than inventing one: the report is the durable C18 run record — its manifest states what the run *was* — while a trace directory is where its output went. The project settled this exact question one cycle earlier for the sibling value, where the unbound-tap names ride beside the report so the CLI, not the library, prints the note (C27/#297). An embedding host gets the handle as a value, never as text to parse back. The chart refusal gains a second arm. A family id (`{campaign8}-{strategy_ordinal}-{instrument}-w..-r..-s..-{run}`) is recognised syntactically — the trailing segment shape plus an 8-hex head, the form `derive_trace_name` mints — and answered with the campaign's REAL recorded handles, filtered to those `chart` would accept, suggesting one only when exactly one matches. It never derives a handle by truncating the id: the id's second segment counts strategies while the handle's counts runs, and one campaign run mints family ids at several strategy ordinals whose traces all live under that single run's handle. `TraceStore::names()` supplies the enumeration, keeping trace file I/O in the store where C22 puts it. Verification caught three errors worth recording. The first spec draft claimed the handle was the id's leading pair — refuted by a green test where a run-0 campaign mints ordinal-1 ids. The second draft's replacement refusal said a campaign run "prints one handle per family"; it prints one per run. Review then found the `NotFound` filter untested — deleting it left the suite green — so the case now has a test that fails without it. closes #309
764 lines
32 KiB
Rust
764 lines
32 KiB
Rust
//! 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); one row at the last warm ts", FoldKind::Count),
|
|
("sum", "sum of the series (f64 taps; f64 row); one row at the last warm ts", FoldKind::Sum),
|
|
(
|
|
"mean",
|
|
"arithmetic mean of the series (f64 taps; f64 row); one row at the last warm ts",
|
|
FoldKind::Mean,
|
|
),
|
|
("min", "minimum of the series (f64 taps; f64 row); one row at the last warm ts", FoldKind::Min),
|
|
("max", "maximum of the series (f64 taps; f64 row); one row at the last warm ts", FoldKind::Max),
|
|
(
|
|
"first",
|
|
"first warm value, at its own timestamp (any kind; kind-preserving row)",
|
|
FoldKind::First,
|
|
),
|
|
("last", "last warm value, at its own timestamp (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 wrap every variant into a returned
|
|
/// `RunnerError` (#297) — prose via `Display`, class via `exit_class`
|
|
/// (content faults 2, store I/O 1, the C14 partition).
|
|
pub enum TapPlanError {
|
|
/// The plan names a tap the blueprint does not declare.
|
|
UnknownTap { name: String, declared: Vec<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, declared } => {
|
|
write!(
|
|
f,
|
|
"the tap plan names '{name}', but the blueprint declares no such tap — declared taps: {}",
|
|
declared.join(", ")
|
|
)
|
|
}
|
|
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)
|
|
}
|
|
}
|
|
|
|
impl TapPlanError {
|
|
/// C14 partition, wired into `run_signal_r`/`run_measurement`'s
|
|
/// `bind_tap_plan` `.map_err` (#297 Fork 1/2): faults in the content of
|
|
/// what argv named are class 2; environment faults (store I/O) are
|
|
/// class 1.
|
|
pub fn exit_class(&self) -> i32 {
|
|
match self {
|
|
TapPlanError::Store(_) => 1,
|
|
TapPlanError::UnknownTap { .. }
|
|
| TapPlanError::UnknownLabel { .. }
|
|
| TapPlanError::KindMismatch { .. }
|
|
| TapPlanError::UnknownParam { .. }
|
|
| TapPlanError::MissingParam { .. }
|
|
| TapPlanError::ParamKind { .. }
|
|
| TapPlanError::Bind(_) => 2,
|
|
}
|
|
}
|
|
}
|
|
|
|
/// 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![],
|
|
doc: "internal recording sink bound to a declared tap",
|
|
}
|
|
}
|
|
|
|
/// 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>>)>,
|
|
/// Declared taps that resolved to no subscription this run (#297): the
|
|
/// caller-printed "unbound" note migrates to the CLI, this is the data
|
|
/// it prints from.
|
|
skipped: Vec<String>,
|
|
/// The trace-store handle this run's taps landed under — `Some` exactly
|
|
/// when the plan persisted something and `begin_run` was called. Rides
|
|
/// beside the report like `skipped` (#297): the CLI prints it, the
|
|
/// library hands it back as a value.
|
|
trace_name: Option<String>,
|
|
}
|
|
|
|
impl BoundTaps {
|
|
/// Every declared tap, declared order (the measurement report's roster).
|
|
pub fn declared_names(&self) -> &[String] {
|
|
&self.declared
|
|
}
|
|
|
|
/// The recorded trace handle, or `None` when this run persisted nothing.
|
|
pub fn trace_name(&self) -> Option<&str> {
|
|
self.trace_name.as_deref()
|
|
}
|
|
|
|
/// Declared taps that resolved to no subscription this run (#297) — the
|
|
/// data the caller's "unbound" note prints from.
|
|
pub fn skipped(&self) -> &[String] {
|
|
&self.skipped
|
|
}
|
|
|
|
/// 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(),
|
|
declared: declared_taps.iter().map(|t| t.name.clone()).collect(),
|
|
});
|
|
}
|
|
}
|
|
|
|
// 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();
|
|
let mut skipped: Vec<String> = 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() }
|
|
}
|
|
// Unbound, inert (C27) — but only reachable when `plan` carries
|
|
// no default (an EXPLICIT plan, e.g. from `--tap`): record-all
|
|
// (`default_named` = `Some(("record", …))`) always resolves the
|
|
// Some arm above, so this arm never fires under record-all and
|
|
// the note is exactly the C14 benign skipped-tap class (#334).
|
|
// The name is recorded here and the note is CLI-printed from
|
|
// the returned `skipped` names (the runner→CLI print
|
|
// migration, #297) — this module no longer emits it.
|
|
None => {
|
|
skipped.push(tap.name.clone());
|
|
continue;
|
|
}
|
|
},
|
|
};
|
|
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(),
|
|
skipped,
|
|
trace_name: if persists { Some(run_name.to_string()) } else { None },
|
|
};
|
|
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");
|
|
}
|
|
|
|
/// C29 entry seam for the registry roster: every entry ships a gate-clean
|
|
/// one-line meaning, and the bind/output prose inside it matches the
|
|
/// entry's executable rules — the drift-pin the retired aura-std
|
|
/// `fold_vocabulary` table carried, moved here with the surface (#332).
|
|
#[test]
|
|
fn roster_docs_pass_the_doc_gate_and_match_the_executable_rules() {
|
|
let r = FoldRegistry::core();
|
|
for entry in r.entries.values() {
|
|
aura_core::doc_gate(entry.label, entry.doc)
|
|
.unwrap_or_else(|f| panic!("fold {} doc fails the gate: {f:?}", entry.label));
|
|
if entry.doc.contains("f64 taps") {
|
|
assert!(
|
|
(entry.binds_at)(ScalarKind::F64) && !(entry.binds_at)(ScalarKind::I64),
|
|
"{} claims f64-only but binds wider",
|
|
entry.label
|
|
);
|
|
} else {
|
|
assert!(
|
|
entry.doc.contains("any kind"),
|
|
"{}: bind prose must be 'f64 taps' or 'any kind': {}",
|
|
entry.label,
|
|
entry.doc
|
|
);
|
|
assert!(
|
|
(entry.binds_at)(ScalarKind::I64) && (entry.binds_at)(ScalarKind::Bool),
|
|
"{} claims any-kind but refuses a kind",
|
|
entry.label
|
|
);
|
|
}
|
|
if entry.doc.contains("i64 row") {
|
|
assert!(
|
|
matches!((entry.output)(ScalarKind::F64), FoldOutput::Row(ScalarKind::I64)),
|
|
"{} claims an i64 row but outputs otherwise",
|
|
entry.label
|
|
);
|
|
} else if entry.doc.contains("f64 row") {
|
|
assert!(
|
|
matches!((entry.output)(ScalarKind::F64), FoldOutput::Row(ScalarKind::F64)),
|
|
"{} claims an f64 row but outputs otherwise",
|
|
entry.label
|
|
);
|
|
} else if entry.doc.contains("kind-preserving row") {
|
|
assert!(
|
|
matches!((entry.output)(ScalarKind::F64), FoldOutput::Row(ScalarKind::F64))
|
|
&& matches!((entry.output)(ScalarKind::I64), FoldOutput::Row(ScalarKind::I64)),
|
|
"{} claims kind-preserving but outputs otherwise",
|
|
entry.label
|
|
);
|
|
} else {
|
|
assert!(
|
|
matches!((entry.output)(ScalarKind::F64), FoldOutput::Series),
|
|
"{}: doc names no row kind, so it must be the series entry",
|
|
entry.label
|
|
);
|
|
}
|
|
}
|
|
}
|
|
|
|
#[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");
|
|
}
|
|
}
|