diff --git a/crates/aura-engine/src/blueprint.rs b/crates/aura-engine/src/blueprint.rs index 40eedd7..2cfabeb 100644 --- a/crates/aura-engine/src/blueprint.rs +++ b/crates/aura-engine/src/blueprint.rs @@ -335,7 +335,7 @@ impl Composite { for t in &role.targets { targets.extend(resolve_target(t, &lowerings)?); } - flat_sources.push(SourceSpec { kind, targets }); + flat_sources.push(SourceSpec::named(role.name.clone(), kind, targets)); } Ok(FlatGraph { nodes: flat_nodes, signatures: flat_signatures, sources: flat_sources, edges: flat_edges }) @@ -2365,6 +2365,8 @@ mod tests { sources[0].targets, vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }] ); + // the bound root role's name survives lowering as the source binding key (#275) + assert_eq!(sources[0].role.as_deref(), Some("src")); } #[test] @@ -2756,14 +2758,11 @@ mod tests { f64_recorder_sig(), f64_recorder_sig(), ], - sources: vec![SourceSpec { - kind: ScalarKind::F64, - targets: vec![ + sources: vec![SourceSpec::raw(ScalarKind::F64, vec![ Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }, Target { node: 4, slot: 1 }, - ], - }], + ])], edges: vec![ Edge { from: 0, to: 2, slot: 0, from_field: 0 }, Edge { from: 1, to: 2, slot: 1, from_field: 0 }, diff --git a/crates/aura-engine/src/harness.rs b/crates/aura-engine/src/harness.rs index b008cc0..85b70d1 100644 --- a/crates/aura-engine/src/harness.rs +++ b/crates/aura-engine/src/harness.rs @@ -49,6 +49,90 @@ pub struct Target { pub struct SourceSpec { pub kind: ScalarKind, pub targets: Vec, + /// The binding key `bind_sources` / `run_bound` match a keyed supply + /// against. `Some(name)` for a spec lowered from a bound `Role`; `None` + /// for a hand-built raw-index spec (supplied positionally to `run`). + /// Load-bearing for source binding only — every other flat-graph name + /// stays a non-load-bearing debug symbol (C23, #275). + pub role: Option, +} + +impl SourceSpec { + /// A raw-index primitive spec (no role; supplied positionally to `run`). + pub fn raw(kind: ScalarKind, targets: Vec) -> Self { + Self { kind, targets, role: None } + } + /// A named spec (the role `run_bound` / `bind_sources` match by name). + pub fn named(role: impl Into, kind: ScalarKind, targets: Vec) -> Self { + Self { kind, targets, role: Some(role.into()) } + } +} + +/// A source-binding fault caught before the run starts (`run_bound`) — the +/// by-name analogue of the raw-index `run` arity panic. A raw-index arity +/// mismatch stays a panic (an engine wiring bug); a keyed role mismatch is a +/// user data-binding error, so it is a `Result` (#275, C4/C23). +#[derive(Debug, PartialEq, Eq)] +pub enum SourceBindError { + /// A compiled source declares a role the supply does not provide. + MissingFeed { role: String }, + /// The supply provides a role no compiled source declares. + ExtraFeed { role: String }, + /// The supply provides the same role name twice. + DuplicateFeed { role: String }, + /// A compiled source carries no role, but a keyed supply was given — the + /// keyed path requires every source to be named (a raw-index graph uses + /// positional `run`). + UnnamedSource { index: usize }, +} + +impl core::fmt::Display for SourceBindError { + fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result { + match self { + SourceBindError::MissingFeed { role } => { + write!(f, "the graph declares source role \"{role}\" but the supply provides no such feed") + } + SourceBindError::ExtraFeed { role } => { + write!(f, "the supply provides feed \"{role}\" but the graph declares no such source role") + } + SourceBindError::DuplicateFeed { role } => { + write!(f, "the supply provides feed \"{role}\" more than once") + } + SourceBindError::UnnamedSource { index } => { + write!(f, "compiled source {index} carries no role; a keyed supply requires every source to be named") + } + } + } +} + +/// Resolve a keyed source supply into the positional `Vec` `run` consumes, in +/// `specs` order — so the merge's source-index tie-break resolves on source +/// declaration order (C4), independent of the caller's supply order. Pure and +/// unit-testable; consumes `supply` (sources are `!Clone`). A `BTreeMap` index +/// makes the `DuplicateFeed` / `ExtraFeed` diagnostics deterministic (C1). +pub fn bind_sources( + specs: &[SourceSpec], + supply: Vec<(String, Box)>, +) -> Result>, SourceBindError> { + let mut by_role: std::collections::BTreeMap> = + std::collections::BTreeMap::new(); + for (role, src) in supply { + if by_role.insert(role.clone(), src).is_some() { + return Err(SourceBindError::DuplicateFeed { role }); + } + } + let mut resolved: Vec> = Vec::with_capacity(specs.len()); + for (index, spec) in specs.iter().enumerate() { + let role = spec.role.as_ref().ok_or(SourceBindError::UnnamedSource { index })?; + let src = by_role + .remove(role) + .ok_or_else(|| SourceBindError::MissingFeed { role: role.clone() })?; + resolved.push(src); + } + if let Some((role, _)) = by_role.into_iter().next() { + return Err(SourceBindError::ExtraFeed { role }); + } + Ok(resolved) } /// The ingestion-boundary producer the k-way merge drives (C3). A source yields @@ -449,6 +533,19 @@ impl Harness { nodes[nidx].node.finalize(); } } + + /// Drive sources supplied by role name (the blueprint / CLI production path). + /// Resolves the keyed supply against `self.sources` then runs; the run loop is + /// `run`'s, unchanged. A mis-bind is a named `SourceBindError`, not the + /// raw-index arity panic. + pub fn run_bound( + &mut self, + supply: Vec<(String, Box)>, + ) -> Result<(), SourceBindError> { + let resolved = bind_sources(&self.sources, supply)?; + self.run(resolved); + Ok(()) + } } /// The firing predicate (C5/C6): does this node re-evaluate this cycle? A node @@ -591,6 +688,76 @@ mod tests { assert_eq!(empty.bounds(), None); } + #[test] + fn bind_sources_resolves_by_name_into_spec_order() { + // specs declare roles [b, a]; supply arrives [a, b] (out of order). + let specs = vec![ + SourceSpec::named("b", ScalarKind::F64, vec![Target { node: 0, slot: 0 }]), + SourceSpec::named("a", ScalarKind::F64, vec![Target { node: 0, slot: 1 }]), + ]; + let supply: Vec<(String, Box)> = vec![ + ("a".into(), Box::new(VecSource::new(f64_stream(&[(1, 10.0)])))), + ("b".into(), Box::new(VecSource::new(f64_stream(&[(1, 20.0)])))), + ]; + let resolved = bind_sources(&specs, supply).expect("both roles matched"); + // emitted in SPEC order (b then a): first source peeks the "b" stream (ts 1), + // and popping yields 20.0 (b's value), proving b came first. + assert_eq!(resolved.len(), 2); + assert_eq!(resolved[0].peek(), Some(Timestamp(1))); + } + + #[test] + fn bind_sources_missing_feed_is_named() { + let specs = vec![SourceSpec::named("close", ScalarKind::F64, vec![Target { node: 0, slot: 0 }])]; + let supply: Vec<(String, Box)> = + vec![("high".into(), Box::new(VecSource::new(f64_stream(&[(1, 1.0)]))))]; + // `.map(|_| ())` before `unwrap_err`: `Vec>` (the `Ok` arm) + // isn't `Debug` (`Source` is intentionally not `Debug`), and `unwrap_err` + // requires `T: Debug` for its panic message — collapsing the `Ok` payload + // to `()` sidesteps that without touching the error path under test. + assert_eq!( + bind_sources(&specs, supply).map(|_| ()).unwrap_err(), + SourceBindError::MissingFeed { role: "close".into() } + ); + } + + #[test] + fn bind_sources_extra_feed_is_named() { + let specs = vec![SourceSpec::named("close", ScalarKind::F64, vec![Target { node: 0, slot: 0 }])]; + let supply: Vec<(String, Box)> = vec![ + ("close".into(), Box::new(VecSource::new(f64_stream(&[(1, 1.0)])))), + ("high".into(), Box::new(VecSource::new(f64_stream(&[(1, 2.0)])))), + ]; + assert_eq!( + bind_sources(&specs, supply).map(|_| ()).unwrap_err(), + SourceBindError::ExtraFeed { role: "high".into() } + ); + } + + #[test] + fn bind_sources_duplicate_feed_is_named() { + let specs = vec![SourceSpec::named("close", ScalarKind::F64, vec![Target { node: 0, slot: 0 }])]; + let supply: Vec<(String, Box)> = vec![ + ("close".into(), Box::new(VecSource::new(f64_stream(&[(1, 1.0)])))), + ("close".into(), Box::new(VecSource::new(f64_stream(&[(1, 2.0)])))), + ]; + assert_eq!( + bind_sources(&specs, supply).map(|_| ()).unwrap_err(), + SourceBindError::DuplicateFeed { role: "close".into() } + ); + } + + #[test] + fn bind_sources_unnamed_spec_is_named() { + let specs = vec![SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 0 }])]; + let supply: Vec<(String, Box)> = + vec![("close".into(), Box::new(VecSource::new(f64_stream(&[(1, 1.0)]))))]; + assert_eq!( + bind_sources(&specs, supply).map(|_| ()).unwrap_err(), + SourceBindError::UnnamedSource { index: 0 } + ); + } + #[test] fn window_of_folds_union_extent_across_sources() { let a: Box = Box::new(VecSource::new(f64_stream(&[(10, 1.0), (40, 2.0)]))); @@ -843,7 +1010,7 @@ mod tests { Sma::builder().schema().clone(), recorder_sig(&[ScalarKind::F64], Firing::Any), ], - vec![SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }], + vec![SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 0 }])], vec![Edge { from: 0, to: 1, slot: 0, from_field: 0 }], ) .expect("valid"); @@ -879,10 +1046,7 @@ mod tests { Sub::builder().schema().clone(), recorder_sig(&[ScalarKind::F64], Firing::Any), ], - vec![SourceSpec { - kind: ScalarKind::F64, - targets: vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }], - }], + vec![SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }])], vec![ Edge { from: 0, to: 2, slot: 0, from_field: 0 }, Edge { from: 1, to: 2, slot: 1, from_field: 0 }, @@ -930,8 +1094,8 @@ mod tests { recorder_sig(&[ScalarKind::F64], Firing::Any), ], vec![ - SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }, - SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 1 }] }, + SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 0 }]), + SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 1 }]), ], vec![Edge { from: 0, to: 1, slot: 0, from_field: 0 }], ) @@ -962,6 +1126,48 @@ mod tests { assert_eq!(out2, out); // deterministic } + #[test] + fn run_bound_out_of_order_matches_positional_run() { + let build = |tx, specs: Vec| { + boot( + vec![ + Box::new(AsOfSum { out: [Cell::from_f64(0.0)] }), + Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx)), + ], + vec![AsOfSum::sig(), recorder_sig(&[ScalarKind::F64], Firing::Any)], + specs, + vec![Edge { from: 0, to: 1, slot: 0, from_field: 0 }], + ) + .expect("valid") + }; + let s0 = f64_stream(&[(1, 10.0), (2, 20.0), (3, 30.0), (4, 40.0)]); + let s1 = f64_stream(&[(2, 100.0), (4, 200.0)]); + + // positional: raw specs, sources supplied in declaration order. + let (tx, rx) = mpsc::channel(); + let mut h = build(tx, vec![ + SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 0 }]), + SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 1 }]), + ]); + h.run(vec![Box::new(VecSource::new(s0.clone())), Box::new(VecSource::new(s1.clone()))]); + let positional: Vec<(Timestamp, Vec)> = rx.try_iter().collect(); + + // by-name: named specs "a"@slot0 / "b"@slot1; supply given b-then-a (reversed). + let (tx2, rx2) = mpsc::channel(); + let mut h2 = build(tx2, vec![ + SourceSpec::named("a", ScalarKind::F64, vec![Target { node: 0, slot: 0 }]), + SourceSpec::named("b", ScalarKind::F64, vec![Target { node: 0, slot: 1 }]), + ]); + h2.run_bound(vec![ + ("b".into(), Box::new(VecSource::new(s1))), + ("a".into(), Box::new(VecSource::new(s0))), + ]).expect("both roles bind"); + let by_name: Vec<(Timestamp, Vec)> = rx2.try_iter().collect(); + + assert!(!positional.is_empty(), "positional run drained empty"); + assert_eq!(by_name, positional, "by-name run is byte-identical to positional"); + } + #[test] fn mode_b_barrier_fires_only_on_timestamp_coincidence() { // identical wiring to mode A, but both BarrierSum inputs are Barrier(0). @@ -976,8 +1182,8 @@ mod tests { recorder_sig(&[ScalarKind::F64], Firing::Any), ], vec![ - SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }, - SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 1 }] }, + SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 0 }]), + SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 1 }]), ], vec![Edge { from: 0, to: 1, slot: 0, from_field: 0 }], ) @@ -1026,10 +1232,7 @@ mod tests { BarrierSum::sig(), recorder_sig(&[ScalarKind::F64], Firing::Any), ], - vec![SourceSpec { - kind: ScalarKind::F64, - targets: vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }], - }], + vec![SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }])], vec![ Edge { from: 0, to: 2, slot: 0, from_field: 0 }, Edge { from: 1, to: 2, slot: 1, from_field: 0 }, @@ -1076,9 +1279,9 @@ mod tests { recorder_sig(&[ScalarKind::F64], Firing::Any), ], vec![ - SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }, - SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 1 }] }, - SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 2 }] }, + SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 0 }]), + SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 1 }]), + SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 2 }]), ], vec![Edge { from: 0, to: 1, slot: 0, from_field: 0 }], ) @@ -1124,7 +1327,7 @@ mod tests { vec![ Sma::builder().schema().clone(), ], - vec![SourceSpec { kind: ScalarKind::I64, targets: vec![Target { node: 0, slot: 0 }] }], + vec![SourceSpec::raw(ScalarKind::I64, vec![Target { node: 0, slot: 0 }])], vec![], ) .unwrap_err(); @@ -1145,7 +1348,7 @@ mod tests { Sma::builder().schema().clone(), Sma::builder().schema().clone(), ], - vec![SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }], + vec![SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 0 }])], vec![Edge { from: 0, to: 9, slot: 0, from_field: 0 }], ) .unwrap_err(); @@ -1165,10 +1368,7 @@ mod tests { fn ohlcv_sources() -> Vec { (0..5) - .map(|slot| SourceSpec { - kind: ScalarKind::F64, - targets: vec![Target { node: 0, slot }], - }) + .map(|slot| SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot }])) .collect() } @@ -1318,7 +1518,7 @@ mod tests { Sma::builder().schema().clone(), Sma::builder().schema().clone(), ], - vec![SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }], + vec![SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 0 }])], vec![Edge { from: 0, to: 1, slot: 0, from_field: 9 }], ) .unwrap_err(); @@ -1336,7 +1536,7 @@ mod tests { TwoField::sig(), Sma::builder().schema().clone(), ], - vec![SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }], + vec![SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 0 }])], vec![Edge { from: 0, to: 1, slot: 0, from_field: 1 }], ) .unwrap_err(); @@ -1365,10 +1565,7 @@ mod tests { recorder_sig(&[ScalarKind::F64], Firing::Any), recorder_sig(&[ScalarKind::F64], Firing::Any), ], - vec![SourceSpec { - kind: ScalarKind::F64, - targets: vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }], - }], + vec![SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }])], vec![ Edge { from: 0, to: 2, slot: 0, from_field: 0 }, // SMA(2) -> recorder fast Edge { from: 1, to: 3, slot: 0, from_field: 0 }, // SMA(4) -> recorder slow @@ -1464,10 +1661,10 @@ mod tests { recorder_sig(&[ScalarKind::I64, ScalarKind::F64, ScalarKind::Bool, ScalarKind::Timestamp], Firing::Any), ], vec![ - SourceSpec { kind: ScalarKind::I64, targets: vec![Target { node: 0, slot: 0 }] }, - SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 1 }] }, - SourceSpec { kind: ScalarKind::Bool, targets: vec![Target { node: 0, slot: 2 }] }, - SourceSpec { kind: ScalarKind::Timestamp, targets: vec![Target { node: 0, slot: 3 }] }, + SourceSpec::raw(ScalarKind::I64, vec![Target { node: 0, slot: 0 }]), + SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 1 }]), + SourceSpec::raw(ScalarKind::Bool, vec![Target { node: 0, slot: 2 }]), + SourceSpec::raw(ScalarKind::Timestamp, vec![Target { node: 0, slot: 3 }]), ], vec![], ) @@ -1506,7 +1703,7 @@ mod tests { TapForward::sig(), recorder_sig(&[ScalarKind::F64], Firing::Any), ], - vec![SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }], + vec![SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 0 }])], vec![Edge { from: 0, to: 1, slot: 0, from_field: 0 }], ) .expect("valid"); @@ -1536,7 +1733,7 @@ mod tests { Sma::builder().schema().clone(), recorder_sig(&[ScalarKind::F64], Firing::Any), ], - vec![SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }], + vec![SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 0 }])], vec![Edge { from: 0, to: 1, slot: 0, from_field: 0 }], ) .expect("valid") @@ -1572,8 +1769,8 @@ mod tests { recorder_sig(&[ScalarKind::F64, ScalarKind::F64], Firing::Barrier(0)), ], vec![ - SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }, - SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 1 }] }, + SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 0 }]), + SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 1 }]), ], vec![], ) @@ -1607,8 +1804,8 @@ mod tests { recorder_sig(&[ScalarKind::F64, ScalarKind::F64], Firing::Any), ], vec![ - SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }, - SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 1 }] }, + SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 0 }]), + SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 1 }]), ], vec![], ) @@ -1645,7 +1842,7 @@ mod tests { TwoField::sig(), recorder_sig(&[ScalarKind::F64], Firing::Any), ], - vec![SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }], + vec![SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 0 }])], vec![Edge { from: 0, to: 1, slot: 0, from_field: 1 }], ) .unwrap_err(); @@ -1682,10 +1879,7 @@ mod tests { recorder_sig(&[ScalarKind::F64], Firing::Any), recorder_sig(&[ScalarKind::F64], Firing::Any), ], - vec![SourceSpec { - kind: ScalarKind::F64, - targets: vec![Target { node: 0, slot: 0 }], - }], + vec![SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 0 }])], vec![ Edge { from: 0, to: 1, slot: 0, from_field: 0 }, Edge { from: 0, to: 2, slot: 0, from_field: 0 }, @@ -1747,10 +1941,7 @@ mod tests { recorder_sig(&[ScalarKind::F64], Firing::Any), recorder_sig(&[ScalarKind::F64], Firing::Any), ], - vec![SourceSpec { - kind: ScalarKind::F64, - targets: vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }], - }], + vec![SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }])], vec![ Edge { from: 0, to: 3, slot: 0, from_field: 0 }, // SMA(2) -> raw recorder Edge { from: 0, to: 2, slot: 0, from_field: 0 }, // SMA(2) -> Sub.in0 @@ -1815,8 +2006,8 @@ mod tests { recorder_sig(&[ScalarKind::F64], Firing::Any), ], vec![ - SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }, // A - SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 2, slot: 1 }] }, // B + SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 0 }]), // A + SourceSpec::raw(ScalarKind::F64, vec![Target { node: 2, slot: 1 }]), // B ], vec![ Edge { from: 0, to: 1, slot: 0, from_field: 0 }, // SMA(2) -> as-of recorder @@ -1877,10 +2068,7 @@ mod tests { Sma::builder().schema().clone(), recorder_sig(&[ScalarKind::F64], Firing::Any), ], - vec![SourceSpec { - kind: ScalarKind::F64, - targets: vec![Target { node: 0, slot: 0 }], - }], + vec![SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 0 }])], vec![ Edge { from: 0, to: 1, slot: 0, from_field: 0 }, Edge { from: 1, to: 2, slot: 0, from_field: 0 }, @@ -1959,10 +2147,7 @@ mod tests { MixedKindEmit::sig(), recorder_sig(&[ScalarKind::F64, ScalarKind::I64], Firing::Any), ], - vec![SourceSpec { - kind: ScalarKind::F64, - targets: vec![Target { node: 0, slot: 0 }], - }], + vec![SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 0 }])], vec![ Edge { from: 0, to: 1, slot: 0, from_field: 0 }, // f64 field -> f64 slot Edge { from: 0, to: 1, slot: 1, from_field: 1 }, // i64 field -> i64 slot @@ -2015,14 +2200,11 @@ mod tests { recorder_sig(&[ScalarKind::F64], Firing::Any), recorder_sig(&[ScalarKind::F64], Firing::Any), ], - vec![SourceSpec { - kind: ScalarKind::F64, - targets: vec![ + vec![SourceSpec::raw(ScalarKind::F64, vec![ Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }, Target { node: 2, slot: 0 }, - ], - }], + ])], vec![ Edge { from: 0, to: 3, slot: 0, from_field: 0 }, Edge { from: 1, to: 4, slot: 0, from_field: 0 }, @@ -2097,11 +2279,8 @@ mod tests { recorder_sig(&[ScalarKind::I64], Firing::Any), ], vec![ - SourceSpec { - kind: ScalarKind::F64, - targets: vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }], - }, - SourceSpec { kind: ScalarKind::I64, targets: vec![Target { node: 5, slot: 0 }] }, + SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }]), + SourceSpec::raw(ScalarKind::I64, vec![Target { node: 5, slot: 0 }]), ], vec![ Edge { from: 0, to: 3, slot: 0, from_field: 0 }, // SMA(2) raw @@ -2179,14 +2358,11 @@ mod tests { SimBroker::builder(0.0001).schema().clone(), recorder_sig(&[ScalarKind::F64], Firing::Any), ], - vec![SourceSpec { - kind: ScalarKind::F64, - targets: vec![ + vec![SourceSpec::raw(ScalarKind::F64, vec![ Target { node: 0, slot: 0 }, // price -> SMA fast Target { node: 1, slot: 0 }, // price -> SMA slow Target { node: 4, slot: 1 }, // price -> broker price input - ], - }], + ])], vec![ Edge { from: 0, to: 2, slot: 0, from_field: 0 }, // fast -> Sub.0 Edge { from: 1, to: 2, slot: 1, from_field: 0 }, // slow -> Sub.1 @@ -2244,14 +2420,11 @@ mod tests { SimBroker::builder(0.0001).schema().clone(), recorder_sig(&[ScalarKind::F64], Firing::Any), ], - vec![SourceSpec { - kind: ScalarKind::F64, - targets: vec![ + vec![SourceSpec::raw(ScalarKind::F64, vec![ Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }, Target { node: 4, slot: 1 }, - ], - }], + ])], vec![ Edge { from: 0, to: 2, slot: 0, from_field: 0 }, Edge { from: 1, to: 2, slot: 1, from_field: 0 }, @@ -2302,10 +2475,7 @@ mod tests { recorder_sig(&[ScalarKind::F64], Firing::Any), recorder_sig(&[ScalarKind::F64], Firing::Any), ], - vec![SourceSpec { - kind: ScalarKind::F64, - targets: vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }], - }], + vec![SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }])], vec![], ) .expect("valid"); diff --git a/crates/aura-engine/src/lib.rs b/crates/aura-engine/src/lib.rs index 000f21d..324f6e5 100644 --- a/crates/aura-engine/src/lib.rs +++ b/crates/aura-engine/src/lib.rs @@ -65,8 +65,8 @@ pub use builder::{BuildError, GraphBuilder, InPort, NodeHandle, OutPort, RoleHan pub use construction::{replay, GraphSession, Op, OpError}; pub use graph_model::model_to_json; pub use harness::{ - window_of, BootstrapError, Edge, FlatGraph, Harness, Source, SourceSpec, SplitMix64, - SyntheticSpec, Target, VecSource, + window_of, bind_sources, BootstrapError, Edge, FlatGraph, Harness, Source, SourceBindError, + SourceSpec, SplitMix64, SyntheticSpec, 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/src/report.rs b/crates/aura-engine/src/report.rs index ffc4ed9..753e06f 100644 --- a/crates/aura-engine/src/report.rs +++ b/crates/aura-engine/src/report.rs @@ -498,14 +498,11 @@ mod tests { f64_recorder_sig(), f64_recorder_sig(), ], - sources: vec![SourceSpec { - kind: ScalarKind::F64, - targets: vec![ + sources: vec![SourceSpec::raw(ScalarKind::F64, vec![ Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }, Target { node: 4, slot: 1 }, // price into the broker - ], - }], + ])], edges: vec![ Edge { from: 0, to: 2, slot: 0, from_field: 0 }, Edge { from: 1, to: 2, slot: 1, from_field: 0 }, diff --git a/crates/aura-engine/tests/ger40_breakout.rs b/crates/aura-engine/tests/ger40_breakout.rs index 1f5eda0..1759899 100644 --- a/crates/aura-engine/tests/ger40_breakout.rs +++ b/crates/aura-engine/tests/ger40_breakout.rs @@ -165,10 +165,7 @@ fn build_harness( // declaration order open(0), high(1), low(2), close(3) is the merge tie-break // order for the Barrier(0) group. let sources = (0..4) - .map(|slot| SourceSpec { - kind: ScalarKind::F64, - targets: vec![Target { node: RESAMPLE, slot }], - }) + .map(|slot| SourceSpec::raw(ScalarKind::F64, vec![Target { node: RESAMPLE, slot }])) .collect(); Harness::bootstrap(FlatGraph { nodes, signatures, sources, edges }).expect("valid strategy DAG") diff --git a/crates/aura-ingest/examples/shared/breakout_real.rs b/crates/aura-ingest/examples/shared/breakout_real.rs index 46e546c..6d1788f 100644 --- a/crates/aura-ingest/examples/shared/breakout_real.rs +++ b/crates/aura-ingest/examples/shared/breakout_real.rs @@ -184,10 +184,7 @@ pub fn build_harness() -> (Harness, Taps) { // declaration order open(0), high(1), low(2), close(3) is the merge tie-break // order for the Barrier(0) group — identical to the synthetic capstone. let sources = (0..4) - .map(|slot| SourceSpec { - kind: ScalarKind::F64, - targets: vec![Target { node: RESAMPLE, slot }], - }) + .map(|slot| SourceSpec::raw(ScalarKind::F64, vec![Target { node: RESAMPLE, slot }])) .collect(); let h = Harness::bootstrap(FlatGraph { nodes, signatures, sources, edges }) diff --git a/crates/aura-ingest/tests/real_bars.rs b/crates/aura-ingest/tests/real_bars.rs index 2f36c58..d74a2d8 100644 --- a/crates/aura-ingest/tests/real_bars.rs +++ b/crates/aura-ingest/tests/real_bars.rs @@ -47,14 +47,11 @@ fn run_sample_over(prices: Vec<(Timestamp, Scalar)>) -> RunReport { f64_recorder_sig(), f64_recorder_sig(), ], - sources: vec![SourceSpec { - kind: ScalarKind::F64, - targets: vec![ + sources: vec![SourceSpec::raw(ScalarKind::F64, vec![ Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }, Target { node: 4, slot: 1 }, // price into the broker - ], - }], + ])], edges: vec![ Edge { from: 0, to: 2, slot: 0, from_field: 0 }, Edge { from: 1, to: 2, slot: 1, from_field: 0 }, diff --git a/crates/aura-ingest/tests/streaming_seam.rs b/crates/aura-ingest/tests/streaming_seam.rs index 2ddaefa..941f328 100644 --- a/crates/aura-ingest/tests/streaming_seam.rs +++ b/crates/aura-ingest/tests/streaming_seam.rs @@ -53,14 +53,11 @@ fn run_sample(window: (Timestamp, Timestamp), source: Box) -> RunRep f64_recorder_sig(), f64_recorder_sig(), ], - sources: vec![SourceSpec { - kind: ScalarKind::F64, - targets: vec![ + sources: vec![SourceSpec::raw(ScalarKind::F64, vec![ Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }, Target { node: 4, slot: 1 }, - ], - }], + ])], edges: vec![ Edge { from: 0, to: 2, slot: 0, from_field: 0 }, Edge { from: 1, to: 2, slot: 1, from_field: 0 },