Brummel/Aura
1
0
Fork 0
Code Issues 33 Pull Requests Actions Packages Projects Releases Wiki Activity

feat(engine): bind ingestion sources by role key — SourceSpec.role + bind_sources/run_bound

Browse Source 
refs #275

The engine-side half of by-name source binding. `SourceSpec` carries a load-bearing
`role: Option<String>` — `Some(name)` for a spec lowered from a bound `Role` (carried
through the blueprint lowering, previously dropped at compile), `None` for a
hand-built raw-index spec. A pure `bind_sources(specs, keyed_supply)` resolves a
`Vec<(role, Box<dyn Source>)>` into the positional `Vec` `run` consumes, in
`SourceSpec` declaration order, erroring named (`SourceBindError`:
Missing/Extra/Duplicate/Unnamed feed) on a mis-bind. `Harness::run_bound` is the
ergonomic method the by-name production path calls.

The raw-index `run(Vec)` primitive and the k-way merge / per-cycle event loop are
untouched: a positional run stays byte-identical (C1, pinned by
`run_bound_out_of_order_matches_positional_run`), and every existing `run(Vec)` caller
keeps working (it ignores `role`). A `BTreeMap` index makes the Duplicate/Extra
diagnostics deterministic.

C4's tie-break guarantee ("source declaration order") is unchanged — `bind_sources`
emits in `SourceSpec` order, so the merge's source-index tie-break resolves on
declaration order regardless of the caller's supply order. The narrow C23 amendment
(a `SourceSpec.role` is load-bearing for source binding, the rest of the flat graph
stays raw-index) is deferred to the ledger note.

The four error-path RED tests assert via `.map(|_| ()).unwrap_err()`: the `Ok` arm
`Vec<Box<dyn Source>>` is not `Debug` (`Source` has no `Debug` supertrait), so a bare
`.unwrap_err()` would not compile — a plan-byte defect corrected here.

Also sweeps the 45 raw-index `SourceSpec { .. }` literals to `SourceSpec::raw(..)`
across the workspace (mechanical, compiler-enumerated) and pins that a bound root
role's name survives lowering.

Verification: `cargo test --workspace` green (0 failed; incl. 5 `bind_sources` cases,
the `run_bound` byte-identity test, the role-survives-lowering pin, and the four
`.sources`-equality twins under the now-stricter `role` equality);
`cargo build --workspace --all-targets` clean; `cargo clippy -p aura-engine
--all-targets` clean.
This commit is contained in:
Claude
2026-07-15 18:46:25 +02:00
parent ad4249f9e6
commit 0b620e1f26
8 changed files with 266 additions and 112 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/aura-engine/src/blueprint.rs
+5 -6
View File
@@ -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 },
crates/aura-engine/src/harness.rs
+251 -81
View File
@@ -49,6 +49,90 @@ pub struct Target {
pub struct SourceSpec {
pub kind: ScalarKind,
pub targets: Vec<Target>,
/// 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<String>,
}
impl SourceSpec {
/// A raw-index primitive spec (no role; supplied positionally to `run`).
pub fn raw(kind: ScalarKind, targets: Vec<Target>) -> Self {
Self { kind, targets, role: None }
}
/// A named spec (the role `run_bound` / `bind_sources` match by name).
pub fn named(role: impl Into<String>, kind: ScalarKind, targets: Vec<Target>) -> 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<dyn Source>)>,
) -> Result<Vec<Box<dyn Source>>, SourceBindError> {
let mut by_role: std::collections::BTreeMap<String, Box<dyn Source>> =
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<Box<dyn Source>> = 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<dyn Source>)>,
) -> 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<dyn Source>)> = 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<dyn Source>)> =
vec![("high".into(), Box::new(VecSource::new(f64_stream(&[(1, 1.0)]))))];
// `.map(|_| ())` before `unwrap_err`: `Vec<Box<dyn Source>>` (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<dyn Source>)> = 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<dyn Source>)> = 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<dyn Source>)> =
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<dyn Source> = 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<SourceSpec>| {
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<Scalar>)> = 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<Scalar>)> = 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<SourceSpec> {
(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");
crates/aura-engine/src/lib.rs
+2 -2
View File
@@ -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,
crates/aura-engine/src/report.rs
+2 -5
View File
@@ -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 },
crates/aura-engine/tests/ger40_breakout.rs
+1 -4
View File
@@ -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")
crates/aura-ingest/examples/shared/breakout_real.rs
+1 -4
View File
@@ -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 })
crates/aura-ingest/tests/real_bars.rs
+2 -5
View File
@@ -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 },
crates/aura-ingest/tests/streaming_seam.rs
+2 -5
View File
@@ -53,14 +53,11 @@ fn run_sample(window: (Timestamp, Timestamp), source: Box<dyn Source>) -> 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 },