From 93d0ec45f2a9dcbc7bd940951d9c4caf0f7684c4 Mon Sep 17 00:00:00 2001 From: claude Date: Fri, 17 Jul 2026 23:37:54 +0200 Subject: [PATCH] feat(engine): FlatGraph::bind_tap + TapBindError (#282) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The run-mode-aware bind seam. bind_tap attaches a CALLER-BUILT Box sink to a declared tap — appending the sink node + an edge from the tap's (node, field) to its input, before bootstrap so the Kahn sort orders it. The engine constructs no aura-std type (it takes a Box), keeping its production dependency aura-core-only. bind_tap raises UndeclaredTap; duplicate detection across binds is the caller's (the engine keeps no cross-call state), mirroring how bind_sources' DuplicateFeed is a supply-level fault. TapBindError is exported for the CLI. A bound tap records exactly the producer's per-cycle output through the full public pipeline; an undeclared name is refused without mutating the graph. refs #282 --- crates/aura-engine/src/harness.rs | 96 ++++++++++++++++++++ crates/aura-engine/src/lib.rs | 2 +- crates/aura-engine/tests/bind_tap_e2e.rs | 111 +++++++++++++++++++++++ 3 files changed, 208 insertions(+), 1 deletion(-) create mode 100644 crates/aura-engine/tests/bind_tap_e2e.rs diff --git a/crates/aura-engine/src/harness.rs b/crates/aura-engine/src/harness.rs index 72cc167..7f4dd3e 100644 --- a/crates/aura-engine/src/harness.rs +++ b/crates/aura-engine/src/harness.rs @@ -312,6 +312,62 @@ pub struct FlatGraph { pub taps: Vec, } +/// A tap-binding fault caught before the run. The output-side twin of +/// `SourceBindError`. `UndeclaredTap` is raised by `bind_tap` itself; +/// `DuplicateBind` is raised by the **caller** (the run-mode-owning layer) when +/// it dedups a tap name across binds — the engine method binds a single tap and +/// keeps no cross-call state, so duplicate detection is the caller's, exactly as +/// `bind_sources`' `DuplicateFeed` is a supply-level, not a per-source, fault. +#[derive(Debug, PartialEq, Eq)] +pub enum TapBindError { + /// A binding names a tap the graph does not declare (raised by `bind_tap`). + UndeclaredTap { name: String }, + /// The same tap name was bound more than once (raised by the caller dedup). + DuplicateBind { name: String }, +} + +impl core::fmt::Display for TapBindError { + fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result { + match self { + TapBindError::UndeclaredTap { name } => { + write!(f, "bind_tap names tap \"{name}\" but the graph declares no such tap") + } + TapBindError::DuplicateBind { name } => { + write!(f, "tap \"{name}\" was bound more than once") + } + } + } +} + +impl FlatGraph { + /// Attach a caller-built sink to a declared tap: append `sink` + `sig` to the + /// flat node/signature arrays and an `Edge` from the tap's `(node, field)` to + /// the sink's input slot 0. Constructs no sink itself (the engine has no + /// `aura-std` production dep) — the caller hands in a `Recorder`. Call BEFORE + /// `bootstrap`; the Kahn sort then orders the appended sink. Binding a tap + /// the graph does not declare is a typed `UndeclaredTap` refusal; duplicate + /// detection across binds is the caller's (this method keeps no cross-call + /// state — see `TapBindError`). + pub fn bind_tap( + &mut self, + tap_name: &str, + sink: Box, + sig: NodeSchema, + ) -> Result<(), TapBindError> { + let tap = self + .taps + .iter() + .find(|t| t.name == tap_name) + .ok_or_else(|| TapBindError::UndeclaredTap { name: tap_name.to_string() })?; + let (from, from_field) = (tap.node, tap.field); + let sink_idx = self.nodes.len(); + self.nodes.push(sink); + self.signatures.push(sig); + self.edges.push(Edge { from, to: sink_idx, slot: 0, from_field }); + Ok(()) + } +} + /// A wiring fault caught once, at bootstrap — C7's "type check paid at wiring" /// generalized to the whole topology. #[derive(Debug, PartialEq, Eq)] @@ -2774,4 +2830,44 @@ mod tests { fired.sort_unstable(); assert_eq!(fired, vec![0, 1]); } + + #[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 + // records exactly the producer's per-cycle output. An undeclared tap name + // is a typed refusal. + let prices = f64_stream(&[(1, 10.0), (2, 20.0), (3, 30.0)]); + // graph: source -> Sma(1) [node 0, producer], declared as tap "d" on (0,0). + let mut flat = FlatGraph { + nodes: vec![Box::new(Sma::new(1))], + signatures: vec![Sma::builder().schema().clone()], + sources: vec![SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 0 }])], + edges: vec![], + taps: vec![FlatTap { name: "d".into(), node: 0, field: 0 }], + }; + // undeclared name refused + let (tx_bad, _rx_bad) = mpsc::channel(); + assert!(matches!( + flat.bind_tap("nope", Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx_bad)), + recorder_sig(&[ScalarKind::F64], Firing::Any)), + Err(TapBindError::UndeclaredTap { .. }) + )); + // declared name binds + let (tx, rx) = mpsc::channel(); + flat.bind_tap("d", Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)), + recorder_sig(&[ScalarKind::F64], Firing::Any)) + .expect("declared tap binds"); + let mut h = Harness::bootstrap(flat).expect("bootstraps with the appended recorder"); + h.run(vec![Box::new(VecSource::new(prices))]); + let recorded: Vec<(Timestamp, Vec)> = rx.try_iter().collect(); + assert_eq!( + recorded, + vec![ + (Timestamp(1), vec![Scalar::f64(10.0)]), + (Timestamp(2), vec![Scalar::f64(20.0)]), + (Timestamp(3), vec![Scalar::f64(30.0)]), + ], + "the bound tap records exactly the producer's per-cycle output" + ); + } } diff --git a/crates/aura-engine/src/lib.rs b/crates/aura-engine/src/lib.rs index 452bfaf..f5fccb3 100644 --- a/crates/aura-engine/src/lib.rs +++ b/crates/aura-engine/src/lib.rs @@ -66,7 +66,7 @@ pub use construction::{replay, GraphSession, Op, OpError}; pub use graph_model::model_to_json; pub use harness::{ window_of, bind_sources, BootstrapError, Edge, FlatGraph, FlatTap, Harness, Source, SourceBindError, - SourceSpec, SplitMix64, SyntheticSpec, Target, VecSource, + SourceSpec, SplitMix64, SyntheticSpec, TapBindError, Target, VecSource, }; pub use report::{ derive_position_events, expected_max_of_normals, f64_field, inv_norm_cdf, join_on_ts, diff --git a/crates/aura-engine/tests/bind_tap_e2e.rs b/crates/aura-engine/tests/bind_tap_e2e.rs new file mode 100644 index 0000000..64bf95f --- /dev/null +++ b/crates/aura-engine/tests/bind_tap_e2e.rs @@ -0,0 +1,111 @@ +//! End-to-end coverage for `FlatGraph::bind_tap` (#282, task 5): the caller-side +//! half of declared taps. `flat_taps_e2e.rs` pins that `compile()` resolves a +//! declared `Tap` into `FlatGraph.taps`; these tests pin the other half — that +//! `bind_tap` actually wires a caller-built sink onto the resolved wire and that +//! it runs through the FULL public pipeline (`Composite::with_taps` -> +//! `compile()` -> `bind_tap` -> `Harness::bootstrap` -> `run`), not just the +//! hand-built `FlatGraph` the in-crate unit test exercises. + +use std::sync::mpsc; + +use aura_engine::{ + Composite, Firing, Harness, NodeSchema, PortSpec, Role, ScalarKind, Scalar, Tap, TapWire, + Target, Timestamp, VecSource, +}; +use aura_std::{Recorder, Sma}; + +/// A single-node composite: one root-bound `price` role feeds `Sma(length=1)` +/// (an identity passthrough), declaring `taps` on its one producer (node 0, +/// field 0). No `OutField` is exported — the tap is the composite's only +/// measurement surface, matching the "declared tap on an interior producer +/// with no consuming edge" shape #282 targets. +fn producer(taps: Vec) -> Composite { + Composite::new( + "producer", + vec![Sma::builder().bind("length", Scalar::i64(1)).into()], + vec![], + vec![Role { name: "price".into(), targets: vec![Target { node: 0, slot: 0 }], source: Some(ScalarKind::F64) }], + vec![], + ) + .with_taps(taps) +} + +/// The `NodeSchema` a `Recorder::new(kinds, firing, tx)` sink satisfies — the +/// caller-side construction contract `bind_tap` requires alongside the boxed +/// node (mirrors `aura_std::Recorder::builder`'s schema, built here directly +/// since `bind_tap` takes a bare `Box`, not a `PrimitiveBuilder`). +fn recorder_sig(kinds: &[ScalarKind]) -> NodeSchema { + NodeSchema { + inputs: kinds + .iter() + .enumerate() + .map(|(i, &kind)| PortSpec { kind, firing: Firing::Any, name: format!("col[{i}]") }) + .collect(), + output: vec![], + params: vec![], + } +} + +fn prices(values: &[f64]) -> Vec<(Timestamp, Scalar)> { + values.iter().enumerate().map(|(i, &v)| (Timestamp(i as i64 + 1), Scalar::f64(v))).collect() +} + +/// Property: **`bind_tap` wires a caller-built sink onto the exact flat +/// producer a declared `Tap` resolved to, all the way through the public +/// `Composite::with_taps -> compile() -> bind_tap -> Harness::bootstrap -> +/// run` pipeline.** The recorded trace must be exactly the tapped producer's +/// per-cycle output — not empty (which a wire pointed at the wrong node/field, +/// or a bind that silently no-ops, would also produce for an unrelated +/// column), and not some other node's output. +#[test] +fn bind_tap_records_the_tapped_producers_output_through_the_public_pipeline() { + let mut flat = producer(vec![Tap { name: "d".into(), from: TapWire { node: 0, field: 0 } }]) + .compile() + .expect("a tap-bearing composite compiles"); + + let (tx, rx) = mpsc::channel(); + flat.bind_tap( + "d", + Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)), + recorder_sig(&[ScalarKind::F64]), + ) + .expect("the declared tap name binds"); + + let mut h = Harness::bootstrap(flat).expect("bootstraps with the appended tap sink"); + h.run(vec![Box::new(VecSource::new(prices(&[10.0, 20.0, 30.0])))]); + + let recorded: Vec = rx.try_iter().map(|(_, row)| row[0].as_f64()).collect(); + assert_eq!( + recorded, + vec![10.0, 20.0, 30.0], + "the bound tap must record exactly the tapped producer's per-cycle output" + ); +} + +/// Property: **`bind_tap` refuses a name the graph did not declare as a tap, +/// and does so without mutating the graph** — the caller can retry with the +/// correct name rather than being left with a half-attached sink or a +/// silently-corrupted node list. `TapBindError` is not part of the crate's +/// public re-export surface (yet), so this asserts the shape observable from +/// outside the crate: an `Err`, and an unmodified flat graph. +#[test] +fn bind_tap_refuses_an_undeclared_tap_name_without_mutating_the_graph() { + let mut flat = producer(vec![Tap { name: "d".into(), from: TapWire { node: 0, field: 0 } }]) + .compile() + .expect("a tap-bearing composite compiles"); + let nodes_before = flat.nodes.len(); + + let (tx, _rx) = mpsc::channel(); + let result = flat.bind_tap( + "nope", + Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)), + recorder_sig(&[ScalarKind::F64]), + ); + + assert!(result.is_err(), "binding an undeclared tap name must be refused"); + assert_eq!( + flat.nodes.len(), + nodes_before, + "a refused bind must not append the sink to the flat node list" + ); +}