feat(engine): firing policies A/B + the k-way ingestion merge

Cycle 0004: the reactive-semantics layer. The engine now merges multiple
timestamped sources into one chronological cycle stream (C3) and gates each
node's re-evaluation by a per-input firing policy (C5/C6) — both rails shipping
together, as the user required ("must sit right from the start").

- `Firing` (aura-core) — a per-input enum on `InputSpec`, where C8 places firing:
  `Any` (mode A, fire-on-any-fresh + hold / as-of join) and `Barrier(u8)` (mode B,
  all-fresh barrier / synchronizing join). Default-free; Sma/Sub declare
  `Firing::Any`, so under a single source ticking every cycle their behavior is
  unchanged (backward compatible — the 0003 fan-out/join DAG still emits
  [None,None,None,2,2,2]).
- `SourceSpec` + the k-way merge (aura-engine) — `run` takes one
  `(Timestamp, Scalar)` stream per source and merges them by (timestamp, source
  index) (C4 ties by declaration order). One record = one cycle. This is the
  permanent ingestion core; the Source trait + data-server IO (C3/C11) are a
  later orthogonal cycle (the user scoped this cycle to the reactive semantics).
- Two read clocks, both load-bearing (resolves the 0003 audit's "0004 owns
  cycle_id" with no write-only field): each push stamps the input slot with
  `fresh_at` (the cycle_id — freshness epoch, C5: fresh iff fresh_at == cycle_id)
  and `last_ts` (the data timestamp — barrier token, C6: a barrier group is
  complete iff all members carry last_ts == T). The `fires()` predicate ORs the
  groups; the ">=1 member fresh this cycle" clause is the once-per-timestamp
  guard. Sample-and-hold falls out of the push model: a non-firing node pushes
  nothing, so consumers keep seeing the held value via window[0].

Key design call, studied against RustAst (src/ast/rtl/streams/): the barrier
token is the data *timestamp*, not cycle_id. RustAst's
last_cycle_per_input.all(== cycle_id) synchronizes one source's fan-out but
cannot synchronize independent sources — under C4 four same-timestamp sources
are four different cycle_ids, so the cycle_id barrier never fires for C6's own
OHLC example. Substituting timestamp is the minimal faithful generalization
(fires both the diamond rejoin and the multi-source bar). Mode A and the k-way
merge are both new (RustAst had neither); RustAst's total_count push counter is
aura's already-built Column::run_count.

Both rails proven on identical sources, firing-only difference:
- mode A (AsOfSum): [None,None,120,130,140,240] — holds the slow input;
- mode B (BarrierSum): [None,None,120,None,None,240] — waits for timestamp
  coincidence;
- mixed A+B (MixedSum): the OR-combine fires both when the barrier pair completes
  (holding the as-of input) and when the as-of input ticks (holding the pair).
Plus determinism (C1, bit-identical re-run of each rail) and the three bootstrap
rejections re-expressed on SourceSpec.

Gates green (re-run by the orchestrator, not just the agent): cargo build/test
(30: aura-core 19 + aura-std 3 + aura-engine 8) / clippy --workspace
--all-targets -D warnings clean; surface-purity grep (no dyn-Any / Rc / RefCell)
clean — the harness doc phrases RustAst's model without the literal tokens to
avoid self-match.

refs walking-skeleton

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

crates/aura-core/src/lib.rs

@@ -23,9 +23,10 @@
 //!   access into each input (closing the cycle-0001 read-side gap on
 //!   [`AnyColumn`]).
 //!
-//! Still to come (subsequent cycles): the firing policies (A: fire-on-any-fresh +
+//! Still to come (subsequent cycles): the `Source` trait + data-server ingestion
-//! hold; B: all-fresh barrier), the deterministic sim loop, sources, and
+//! and source-native time normalization (C3/C11), the broker-independent
-//! ingestion.
+//! position-event output and downstream broker nodes (C10), and the run registry
+//! (C18/C22).
 
 mod any;
 mod column;
@@ -38,5 +39,5 @@ pub use any::AnyColumn;
 pub use column::{Column, Window};
 pub use ctx::Ctx;
 pub use error::KindMismatch;
-pub use node::{InputSpec, Node, NodeSchema};
+pub use node::{Firing, InputSpec, Node, NodeSchema};
 pub use scalar::{Scalar, ScalarKind, Timestamp};

crates/aura-core/src/node.rs

@@ -1,16 +1,32 @@
 //! The node contract (C8): the interface every node implements. A node declares
-//! its inputs and output kind via `schema`, and computes one cycle's output via
+//! its inputs (each with its scalar kind, lookback depth, and firing policy, C6)
-//! `eval`. Firing policy (C6) and tunable params (C12/C19) are deliberately not
+//! and its output kind via `schema`, and computes one cycle's output via `eval`.
-//! part of the schema yet — see spec 0002's "Out of scope".
+//! Tunable params (C12/C19) are deliberately not part of the schema yet — see
+//! spec 0002's "Out of scope".
 
 use crate::{Ctx, Scalar, ScalarKind};
 
-/// One declared input of a node: its scalar kind and the lookback depth the
+/// The firing policy of one input (C6): how the engine decides whether a node
-/// engine must pre-size for it (must be >= 1).
+/// re-evaluates this cycle. A node fires when *any* of its input groups fires
+/// (OR over groups).
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum Firing {
+    /// Mode A — fire-on-any-fresh + hold (as-of join): fires the node whenever
+    /// this input is fresh this cycle; a stale input contributes its held value.
+    Any,
+    /// Mode B — all-fresh barrier (synchronizing join): this input is a member of
+    /// barrier group `N`; the group fires the node only when every member shares
+    /// the current cycle's timestamp.
+    Barrier(u8),
+}
+
+/// One declared input of a node: its scalar kind, the lookback depth the engine
+/// must pre-size for it (must be >= 1), and its firing policy (C6).
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 pub struct InputSpec {
     pub kind: ScalarKind,
     pub lookback: usize,
+    pub firing: Firing,
 }
 
 /// A node's declared interface: its inputs (in order) and its single output
@@ -29,3 +45,17 @@ pub trait Node {
     fn schema(&self) -> NodeSchema;
     fn eval(&mut self, ctx: Ctx<'_>) -> Option<Scalar>;
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn input_spec_carries_firing() {
+        let a = InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Any };
+        let b = InputSpec { kind: ScalarKind::F64, lookback: 2, firing: Firing::Barrier(0) };
+        assert_eq!(a.firing, Firing::Any);
+        assert_eq!(b.firing, Firing::Barrier(0));
+        assert_ne!(a.firing, b.firing);
+    }
+}

crates/aura-engine/src/harness.rs

@@ -1,16 +1,24 @@
 //! The harness — the closed root graph that runs (glossary: `harness`, C20) —
-//! and its deterministic single-source run loop. A `Harness` is a bootstrapped,
+//! and its deterministic run loop. A `Harness` is a bootstrapped, frozen root
-//! frozen root graph: a flat node array plus an index edge table, topologically
+//! graph: a flat node array plus an index edge table, topologically ordered,
-//! ordered, driven cycle by cycle by one source. It is the flat, monomorphized
+//! driven cycle by cycle by a k-way merge of timestamped sources (C3/C4). It is
-//! sharpening of RustAst's reference-counted, interior-mutable observer push
+//! the flat, monomorphized sharpening of RustAst's reference-counted,
-//! graph: no reference counting, no interior mutability, no per-cycle allocation
+//! interior-mutable observer push graph: no reference counting, no interior
-//! (C1/C7). Each node owns its input columns (the cycle-0002 shape), so `Ctx` is
+//! mutability, no per-cycle allocation (C1/C7). Each node owns its input columns
-//! unchanged; a producer's `eval`
+//! (the cycle-0002 shape), so `Ctx` is unchanged.
-//! output is forwarded into its consumers' input columns, and a `None` forwards
+//!
-//! nothing (the structural seed of sample-and-hold, realized with freshness in a
+//! Firing (C5/C6) gates re-evaluation. Two read clocks drive it, both stamped per
-//! later cycle).
+//! input slot on every push: `fresh_at` (the `cycle_id` of the last push —
+//! freshness epoch, C5) and `last_ts` (the data timestamp of that push — barrier
+//! token, C6). A node fires when any `Firing::Any` input is fresh this cycle, or
+//! when a `Firing::Barrier` group has every member at the current timestamp; a
+//! node that does not fire pushes nothing, so consumers keep seeing the held
+//! value via `window[0]` (sample-and-hold falls out of the push model). The
+//! barrier token is the timestamp, not the `cycle_id`: under C4 four same-time
+//! sources are four cycles, so RustAst's `cycle_id`-equality barrier could never
+//! fire across sources — the timestamp generalizes it faithfully.
 
-use aura_core::{AnyColumn, Ctx, Node, Scalar, ScalarKind, Timestamp};
+use aura_core::{AnyColumn, Ctx, Firing, Node, Scalar, ScalarKind, Timestamp};
 
 /// A producer-output -> consumer-input-slot forwarding edge.
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
@@ -27,6 +35,16 @@ pub struct Target {
     pub slot: usize,
 }
 
+/// A declared source: the scalar kind it produces and the input slots each of
+/// its records is forwarded into. The multi-source generalization of cycle
+/// 0003's `(source_targets, source_kind)` pair — sources are k-way-merged by
+/// timestamp at ingestion (C3); there is no merge inside the graph.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct SourceSpec {
+    pub kind: ScalarKind,
+    pub targets: Vec<Target>,
+}
+
 /// A wiring fault caught once, at bootstrap — C7's "type check paid at wiring"
 /// generalized to the whole topology.
 #[derive(Debug, PartialEq, Eq)]
@@ -40,9 +58,22 @@ pub enum BootstrapError {
     Cycle,
 }
 
+/// Per-input engine bookkeeping (invisible to nodes): the two read clocks of the
+/// firing machinery. `fresh_at` is the `cycle_id` of the last push into this slot
+/// (freshness epoch, C5: fresh this cycle iff `fresh_at == cycle_id`); `last_ts`
+/// is the data timestamp of that push (barrier token, C6: a barrier group is
+/// complete iff all members carry `last_ts == T`). A never-pushed slot keeps the
+/// cold sentinel `Timestamp(i64::MIN)`, which no real timestamp equals.
+struct SlotState {
+    fresh_at: u64,
+    last_ts: Timestamp,
+}
+
 struct NodeBox {
     node: Box<dyn Node>,
     inputs: Vec<AnyColumn>,
+    firing: Vec<Firing>,
+    slots: Vec<SlotState>,
 }
 
 /// A bootstrapped, frozen root graph instance plus its deterministic run loop.
@@ -50,7 +81,7 @@ pub struct Harness {
     nodes: Vec<NodeBox>,
     topo: Vec<usize>,
     out_edges: Vec<Vec<Edge>>,
-    source_targets: Vec<Target>,
+    sources: Vec<SourceSpec>,
     observe: usize,
 }
 
@@ -64,7 +95,7 @@ impl core::fmt::Debug for Harness {
             .field("nodes", &self.nodes.len())
             .field("topo", &self.topo)
             .field("out_edges", &self.out_edges)
-            .field("source_targets", &self.source_targets)
+            .field("sources", &self.sources)
             .field("observe", &self.observe)
             .finish()
     }
@@ -72,13 +103,13 @@ impl core::fmt::Debug for Harness {
 
 impl Harness {
     /// Bind nodes + wiring into a frozen, runnable graph. Sizes each node's input
-    /// columns from its `schema`, kind-checks every edge and source target, and
+    /// columns from its `schema`, lifts each input's firing policy, initializes
+    /// per-slot freshness state, kind-checks every source target and edge, and
     /// topologically orders the nodes (Kahn), rejecting any directed cycle.
     pub fn bootstrap(
         nodes: Vec<Box<dyn Node>>,
-        source_targets: Vec<Target>,
+        sources: Vec<SourceSpec>,
         edges: Vec<Edge>,
-        source_kind: ScalarKind,
         observe: usize,
     ) -> Result<Harness, BootstrapError> {
         let n = nodes.len();
@@ -88,7 +119,7 @@ impl Harness {
 
         let schemas: Vec<_> = nodes.iter().map(|nd| nd.schema()).collect();
 
-        // size each node's own input columns from its schema
+        // size each node's input columns from its schema; lift firing; init slots
         let mut boxes: Vec<NodeBox> = Vec::with_capacity(n);
         for (nd, schema) in nodes.into_iter().zip(schemas.iter()) {
             let inputs: Vec<AnyColumn> = schema
@@ -96,18 +127,26 @@ impl Harness {
                 .iter()
                 .map(|spec| AnyColumn::with_capacity(spec.kind, spec.lookback))
                 .collect();
-            boxes.push(NodeBox { node: nd, inputs });
+            let firing: Vec<Firing> = schema.inputs.iter().map(|spec| spec.firing).collect();
+            let slots: Vec<SlotState> = schema
+                .inputs
+                .iter()
+                .map(|_| SlotState { fresh_at: 0, last_ts: Timestamp(i64::MIN) })
+                .collect();
+            boxes.push(NodeBox { node: nd, inputs, firing, slots });
         }
 
-        // source targets: the source value must match each target slot's kind
+        // source targets: each source's value must match each of its target slots' kind
-        for t in &source_targets {
+        for src in &sources {
-            let s = schemas.get(t.node).ok_or(BootstrapError::BadIndex)?;
+            for t in &src.targets {
-            let slot = s.inputs.get(t.slot).ok_or(BootstrapError::BadIndex)?;
+                let s = schemas.get(t.node).ok_or(BootstrapError::BadIndex)?;
-            if slot.kind != source_kind {
+                let slot = s.inputs.get(t.slot).ok_or(BootstrapError::BadIndex)?;
-                return Err(BootstrapError::KindMismatch {
+                if slot.kind != src.kind {
-                    producer: source_kind,
+                    return Err(BootstrapError::KindMismatch {
-                    consumer: slot.kind,
+                        producer: src.kind,
-                });
+                        consumer: slot.kind,
+                    });
+                }
             }
         }
 
@@ -153,34 +192,74 @@ impl Harness {
             nodes: boxes,
             topo,
             out_edges,
-            source_targets,
+            sources,
             observe,
         })
     }
 
-    /// Drive the records in timestamp order; returns the observed node's per-cycle
+    /// Drive the sources, k-way-merged in timestamp order (ties by source index,
-    /// output. Allocates nothing on the per-cycle eval/forward path.
+    /// C4); returns the observed node's per-cycle emission (`Some` when it fired
-    pub fn run(
+    /// and produced output, `None` when it held or filtered). One stream per
-        &mut self,
+    /// source, each ascending in timestamp (C3 ingestion precondition). Allocates
-        records: impl Iterator<Item = (Timestamp, Scalar)>,
+    /// nothing per cycle beyond the output vector.
-    ) -> Vec<Option<Scalar>> {
+    pub fn run(&mut self, streams: Vec<Vec<(Timestamp, Scalar)>>) -> Vec<Option<Scalar>> {
-        // disjoint field borrows so the topo walk can read `topo`/`out_edges`
+        assert_eq!(
-        // while mutating `nodes`
+            streams.len(),
-        let Harness { nodes, topo, out_edges, source_targets, observe } = self;
+            self.sources.len(),
+            "run: one stream per source required (got {} streams for {} sources)",
+            streams.len(),
+            self.sources.len()
+        );
+
+        // disjoint field borrows so the topo walk can read topo/out_edges/sources
+        // while mutating nodes
+        let Harness { nodes, topo, out_edges, sources, observe } = self;
         let observe = *observe;
 
+        let mut cursor: Vec<usize> = vec![0; streams.len()];
+        let mut cycle_id: u64 = 0;
         let mut out = Vec::new();
-        for (_ts, value) in records {
+
-            // forward the source value into its target input slots
+        loop {
-            for t in source_targets.iter() {
+            // pick the live source head with the smallest (timestamp, source index)
-                nodes[t.node].inputs[t.slot]
+            let mut pick: Option<usize> = None;
-                    .push(value)
+            for (s, stream) in streams.iter().enumerate() {
-                    .expect("source kind checked at wiring");
+                if cursor[s] < stream.len() {
+                    match pick {
+                        None => pick = Some(s),
+                        Some(p) => {
+                            if stream[cursor[s]].0 < streams[p][cursor[p]].0 {
+                                pick = Some(s);
+                            }
+                        }
+                    }
+                }
+            }
+            let s = match pick {
+                Some(s) => s,
+                None => break, // all streams exhausted
+            };
+            let (ts, value) = streams[s][cursor[s]];
+            cursor[s] += 1;
+            cycle_id += 1;
+
+            // forward the source value into its target slots, stamping freshness
+            for t in sources[s].targets.iter() {
+                let nb = &mut nodes[t.node];
+                nb.inputs[t.slot].push(value).expect("source kind checked at wiring");
+                nb.slots[t.slot] = SlotState { fresh_at: cycle_id, last_ts: ts };
             }
 
-            // evaluate in topological order; capture the observed output; forward
+            // evaluate in topological order; gate by firing; forward Some outputs
             let mut observed = None;
             for &nidx in topo.iter() {
+                let fired = {
+                    let nb = &nodes[nidx];
+                    fires(&nb.firing, &nb.slots, cycle_id, ts)
+                };
+                if !fired {
+                    continue; // hold: no eval, no push
+                }
                 let result = {
                     let nb = &mut nodes[nidx];
                     nb.node.eval(Ctx::new(&nb.inputs))
@@ -190,9 +269,9 @@ impl Harness {
                 }
                 if let Some(v) = result {
                     for e in out_edges[nidx].iter() {
-                        nodes[e.to].inputs[e.slot]
+                        let nb = &mut nodes[e.to];
-                            .push(v)
+                        nb.inputs[e.slot].push(v).expect("edge kind checked at wiring");
-                            .expect("edge kind checked at wiring");
+                        nb.slots[e.slot] = SlotState { fresh_at: cycle_id, last_ts: ts };
                     }
                 }
             }
@@ -202,31 +281,148 @@ impl Harness {
     }
 }
 
+/// The firing predicate (C5/C6): does this node re-evaluate this cycle? A node
+/// fires when *any* of its input groups fires (OR). A `Firing::Any` input fires
+/// the node when it is fresh this cycle (`fresh_at == cycle_id`). A barrier group
+/// fires when >=1 member is fresh this cycle AND every member carries
+/// `last_ts == ts` — the ">=1 fresh" clause is the once-per-timestamp guard (a
+/// group completed in a prior cycle has no fresh member now, so it does not
+/// re-fire).
+fn fires(firing: &[Firing], slots: &[SlotState], cycle_id: u64, ts: Timestamp) -> bool {
+    // mode A: any fire-on-any-fresh input that is fresh this cycle
+    for (i, f) in firing.iter().enumerate() {
+        if matches!(f, Firing::Any) && slots[i].fresh_at == cycle_id {
+            return true;
+        }
+    }
+    // mode B: each distinct barrier group fires when complete this cycle
+    if let Some(max_group) = firing.iter().filter_map(group_id).max() {
+        for g in 0..=max_group {
+            let mut has_member = false;
+            let mut any_fresh = false;
+            let mut all_at_ts = true;
+            for (i, f) in firing.iter().enumerate() {
+                if group_id(f) == Some(g) {
+                    has_member = true;
+                    if slots[i].fresh_at == cycle_id {
+                        any_fresh = true;
+                    }
+                    if slots[i].last_ts != ts {
+                        all_at_ts = false;
+                    }
+                }
+            }
+            if has_member && any_fresh && all_at_ts {
+                return true;
+            }
+        }
+    }
+    false
+}
+
+/// The barrier group id of a firing policy, or `None` for mode A.
+fn group_id(f: &Firing) -> Option<u8> {
+    match f {
+        Firing::Any => None,
+        Firing::Barrier(g) => Some(*g),
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
+    // InputSpec / NodeSchema are not imported by harness.rs production code (it
+    // only reads `nd.schema()` fields, never naming the types), so they are not
+    // brought in by `use super::*` — the fixtures construct them, so import here.
+    use aura_core::{InputSpec, NodeSchema};
     use aura_std::{Sma, Sub};
 
-    fn f64_records(prices: &[f64]) -> impl Iterator<Item = (Timestamp, Scalar)> + '_ {
+    /// Build an f64 source stream from (timestamp, value) points.
-        prices
+    fn f64_stream(points: &[(i64, f64)]) -> Vec<(Timestamp, Scalar)> {
-            .iter()
+        points.iter().map(|&(t, v)| (Timestamp(t), Scalar::F64(v))).collect()
-            .enumerate()
+    }
-            .map(|(i, &p)| (Timestamp(i as i64), Scalar::F64(p)))
+
+    // --- firing-policy fixtures (test-local; not library nodes — C9: examples
+    // for the engine's own tests, no speculative aura-std surface) ---
+
+    /// Mode A as-of join: a 2-input f64 sum that fires whenever either input is
+    /// fresh, holding the other. Warm-up returns None until both have a value.
+    struct AsOfSum;
+    impl Node for AsOfSum {
+        fn schema(&self) -> NodeSchema {
+            NodeSchema {
+                inputs: vec![
+                    InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Any },
+                    InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Any },
+                ],
+                output: ScalarKind::F64,
+            }
+        }
+        fn eval(&mut self, ctx: Ctx<'_>) -> Option<Scalar> {
+            let a = ctx.f64_in(0);
+            let b = ctx.f64_in(1);
+            if a.is_empty() || b.is_empty() {
+                return None;
+            }
+            Some(Scalar::F64(a[0] + b[0]))
+        }
+    }
+
+    /// Mode B barrier join: a 2-input f64 sum that fires only when both inputs
+    /// share the current cycle timestamp (both warm by construction when it fires).
+    struct BarrierSum;
+    impl Node for BarrierSum {
+        fn schema(&self) -> NodeSchema {
+            NodeSchema {
+                inputs: vec![
+                    InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Barrier(0) },
+                    InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Barrier(0) },
+                ],
+                output: ScalarKind::F64,
+            }
+        }
+        fn eval(&mut self, ctx: Ctx<'_>) -> Option<Scalar> {
+            Some(Scalar::F64(ctx.f64_in(0)[0] + ctx.f64_in(1)[0]))
+        }
+    }
+
+    /// Mixed A+B: barrier pair (inputs 0,1 in group 0) plus an as-of input
+    /// (input 2). Fires when the pair completes (holding input 2) OR when input 2
+    /// ticks (holding the pair) — the OR-combine.
+    struct MixedSum;
+    impl Node for MixedSum {
+        fn schema(&self) -> NodeSchema {
+            NodeSchema {
+                inputs: vec![
+                    InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Barrier(0) },
+                    InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Barrier(0) },
+                    InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Any },
+                ],
+                output: ScalarKind::F64,
+            }
+        }
+        fn eval(&mut self, ctx: Ctx<'_>) -> Option<Scalar> {
+            let a = ctx.f64_in(0);
+            let b = ctx.f64_in(1);
+            let c = ctx.f64_in(2);
+            if a.is_empty() || b.is_empty() || c.is_empty() {
+                return None;
+            }
+            Some(Scalar::F64(a[0] + b[0] + c[0]))
+        }
     }
 
     #[test]
     fn chain_source_sma_runs() {
-        // node 0 = SMA(3); source -> SMA(3).in0; observe node 0
+        // node 0 = SMA(3); one source -> SMA(3).in0; observe node 0
-        let nodes: Vec<Box<dyn Node>> = vec![Box::new(Sma::new(3))];
+        let mut h = Harness::bootstrap(
-        let mut sim = Harness::bootstrap(
+            vec![Box::new(Sma::new(3))],
-            nodes,
+            vec![SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }],
-            vec![Target { node: 0, slot: 0 }],
             vec![],
-            ScalarKind::F64,
             0,
         )
         .expect("valid");
-        let out = sim.run(f64_records(&[1.0, 2.0, 3.0, 4.0, 5.0]));
+        let out = h.run(vec![f64_stream(&[(1, 1.0), (2, 2.0), (3, 3.0), (4, 4.0), (5, 5.0)])]);
         assert_eq!(
             out,
             vec![
@@ -241,25 +437,25 @@ mod tests {
 
     #[test]
     fn fan_out_join_dag_runs_deterministically() {
-        // 0 = SMA(2), 1 = SMA(4), 2 = Sub; source fans into both SMAs; SMAs join into Sub
+        // 0 = SMA(2), 1 = SMA(4), 2 = Sub; one source fans into both SMAs (all
-        let build = || -> Harness {
+        // Any), SMAs join into Sub — the 0003 compat baseline on the new API.
-            let nodes: Vec<Box<dyn Node>> =
+        let build = || {
-                vec![Box::new(Sma::new(2)), Box::new(Sma::new(4)), Box::new(Sub::new())];
             Harness::bootstrap(
-                nodes,
+                vec![Box::new(Sma::new(2)), Box::new(Sma::new(4)), Box::new(Sub::new())],
-                vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }],
+                vec![SourceSpec {
+                    kind: ScalarKind::F64,
+                    targets: vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }],
+                }],
                 vec![Edge { from: 0, to: 2, slot: 0 }, Edge { from: 1, to: 2, slot: 1 }],
-                ScalarKind::F64,
                 2,
             )
             .expect("valid DAG")
         };
+        let prices = f64_stream(&[(1, 10.0), (2, 12.0), (3, 14.0), (4, 16.0), (5, 18.0), (6, 20.0)]);
 
-        let prices = [10.0_f64, 12.0, 14.0, 16.0, 18.0, 20.0];
+        let mut h = build();
-
+        let out = h.run(vec![prices.clone()]);
-        let mut sim = build();
+        // Sub fires once SMA(4) is warm (cycle 4): 15-13, 17-15, 19-17 -> 2.
-        let out = sim.run(f64_records(&prices));
-        // Sub fires only once SMA(4) is warm (cycle index 3): 15-13, 17-15, 19-17 -> 2.
         assert_eq!(
             out,
             vec![
@@ -273,20 +469,126 @@ mod tests {
         );
 
         // determinism (C1): a second identical run is bit-identical
-        let mut sim2 = build();
+        let mut h2 = build();
-        let out2 = sim2.run(f64_records(&prices));
+        assert_eq!(h2.run(vec![prices]), out);
-        assert_eq!(out, out2);
+    }
+
+    #[test]
+    fn mode_a_as_of_fires_on_any_fresh_and_holds() {
+        // source 0 ticks t=1,2,3,4; source 1 ticks t=2,4 (slower); both inputs Any.
+        let build = || {
+            Harness::bootstrap(
+                vec![Box::new(AsOfSum)],
+                vec![
+                    SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] },
+                    SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 1 }] },
+                ],
+                vec![],
+                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)]);
+
+        let mut h = build();
+        let out = h.run(vec![s0.clone(), s1.clone()]);
+        // holds s1=100 across t=3 and the t=4 s0-cycle; emits on every tick once warm.
+        assert_eq!(
+            out,
+            vec![
+                None,
+                None,
+                Some(Scalar::F64(120.0)),
+                Some(Scalar::F64(130.0)),
+                Some(Scalar::F64(140.0)),
+                Some(Scalar::F64(240.0)),
+            ]
+        );
+
+        let mut h2 = build();
+        assert_eq!(h2.run(vec![s0, s1]), out); // deterministic
+    }
+
+    #[test]
+    fn mode_b_barrier_fires_only_on_timestamp_coincidence() {
+        // identical wiring to mode A, but both inputs are Barrier(0).
+        let build = || {
+            Harness::bootstrap(
+                vec![Box::new(BarrierSum)],
+                vec![
+                    SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] },
+                    SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 1 }] },
+                ],
+                vec![],
+                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)]);
+
+        let mut h = build();
+        let out = h.run(vec![s0.clone(), s1.clone()]);
+        // emits ONLY at t=2 and t=4 where both inputs share the timestamp; holds otherwise.
+        assert_eq!(
+            out,
+            vec![
+                None,
+                None,
+                Some(Scalar::F64(120.0)),
+                None,
+                None,
+                Some(Scalar::F64(240.0)),
+            ]
+        );
+
+        let mut h2 = build();
+        assert_eq!(h2.run(vec![s0, s1]), out); // deterministic
+    }
+
+    #[test]
+    fn mixed_a_and_b_or_combine_on_one_node() {
+        // in0,in1 = barrier group 0 (sources 0,1); in2 = as-of (source 2).
+        let mut h = Harness::bootstrap(
+            vec![Box::new(MixedSum)],
+            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 }] },
+            ],
+            vec![],
+            0,
+        )
+        .expect("valid");
+        let s0 = f64_stream(&[(2, 20.0), (5, 50.0)]); // in0 (barrier)
+        let s1 = f64_stream(&[(2, 200.0)]);           // in1 (barrier)
+        let s2 = f64_stream(&[(1, 1.0), (3, 3.0)]);   // in2 (as-of)
+
+        let out = h.run(vec![s0, s1, s2]);
+        // cycle order (ts, source idx): (1,s2) (2,s0) (2,s1) (3,s2) (5,s0).
+        // c3: barrier pair completes at t=2, holds c=1 -> 20+200+1 = 221.
+        // c4: as-of input ticks at t=3, holds the pair -> 20+200+3 = 223.
+        // c1 fires on the as-of input but filters (pair cold); c2,c5 hold.
+        assert_eq!(
+            out,
+            vec![
+                None,
+                None,
+                Some(Scalar::F64(221.0)),
+                Some(Scalar::F64(223.0)),
+                None,
+            ]
+        );
     }
 
     #[test]
     fn bootstrap_rejects_a_cycle() {
         // two SMA(1) nodes wired a -> b -> a
-        let nodes: Vec<Box<dyn Node>> = vec![Box::new(Sma::new(1)), Box::new(Sma::new(1))];
         let err = Harness::bootstrap(
-            nodes,
+            vec![Box::new(Sma::new(1)), Box::new(Sma::new(1))],
             vec![],
             vec![Edge { from: 0, to: 1, slot: 0 }, Edge { from: 1, to: 0, slot: 0 }],
-            ScalarKind::F64,
             0,
         )
         .unwrap_err();
@@ -295,13 +597,11 @@ mod tests {
 
     #[test]
     fn bootstrap_rejects_a_kind_mismatch() {
-        // SMA(1) declares an f64 input; feeding an i64 source mismatches
+        // SMA(1) declares an f64 input; an i64 source mismatches
-        let nodes: Vec<Box<dyn Node>> = vec![Box::new(Sma::new(1))];
         let err = Harness::bootstrap(
-            nodes,
+            vec![Box::new(Sma::new(1))],
-            vec![Target { node: 0, slot: 0 }],
+            vec![SourceSpec { kind: ScalarKind::I64, targets: vec![Target { node: 0, slot: 0 }] }],
             vec![],
-            ScalarKind::I64,
             0,
         )
         .unwrap_err();
@@ -314,12 +614,10 @@ mod tests {
     #[test]
     fn bootstrap_rejects_a_bad_index() {
         // observe node 5 does not exist
-        let nodes: Vec<Box<dyn Node>> = vec![Box::new(Sma::new(1))];
         let err = Harness::bootstrap(
-            nodes,
+            vec![Box::new(Sma::new(1))],
-            vec![Target { node: 0, slot: 0 }],
+            vec![SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }],
             vec![],
-            ScalarKind::F64,
             5,
         )
         .unwrap_err();

crates/aura-engine/src/lib.rs

@@ -1,24 +1,26 @@
 //! `aura-engine` — the headless, UI-agnostic reactive SoA engine.
 //!
-//! Delivered in cycle 0003 — the harness and its deterministic single-source
+//! Delivered across cycles 0003-0004 — the harness and its deterministic run
-//! run loop:
+//! loop:
 //!
 //! - [`Harness`] — the closed root graph that runs (a flat node array + an index
-//!   edge table, topologically ordered) and its deterministic `run` loop: one
+//!   edge table, topologically ordered) and its deterministic `run` loop: a
-//!   source driven through a wired DAG of nodes, cycle by cycle, each output
+//!   k-way merge of timestamped sources (C3/C4) driven through a wired DAG of
-//!   forwarded into its consumers' input columns (C1/C4/C7/C8/C9).
+//!   nodes, cycle by cycle, with freshness-gated recompute and the two firing
-//! - [`Edge`] / [`Target`] — producer->consumer and source->consumer wiring.
+//!   policies (C5/C6) deciding when each node re-evaluates and what it holds.
+//! - [`Edge`] / [`Target`] / [`SourceSpec`] — producer->consumer wiring,
+//!   source->consumer wiring, and a declared source (its kind + target slots).
 //! - [`BootstrapError`] — wiring faults caught once, at bootstrap (kind
 //!   mismatch, bad index, directed cycle).
 //!
-//! Still to come (subsequent cycles): freshness-gated recompute / sample-and-hold
+//! Still to come (subsequent cycles): the `Source` trait + data-server ingestion
-//! (C5), the ingestion boundary (k-way merge of timestamped sources, C3), the
+//! and source-native time normalization (C3/C11), the broker-independent
-//! broker-independent position-event output and downstream broker nodes (C10),
+//! position-event output and downstream broker nodes (C10), and the atomic sim
-//! and the atomic sim unit `(topology + params + data-window + seed) -> metrics`
+//! unit `(topology + params + data-window + seed) -> metrics` that the sweep /
-//! that the sweep / optimize / walk-forward / Monte-Carlo axes orchestrate.
+//! optimize / walk-forward / Monte-Carlo axes orchestrate.
 //!
 //! Visualization is never here: it is a downstream consumer node on the streams.
 
 mod harness;
 
-pub use harness::{BootstrapError, Edge, Harness, Target};
+pub use harness::{BootstrapError, Edge, Harness, SourceSpec, Target};

crates/aura-std/src/sma.rs

@@ -3,7 +3,7 @@
 //! `Node` contract is authorable from a downstream crate and evaluable with no
 //! engine present (the test drives it by hand, as the sim loop later will).
 
-use aura_core::{Ctx, InputSpec, Node, NodeSchema, Scalar, ScalarKind};
+use aura_core::{Ctx, Firing, InputSpec, Node, NodeSchema, Scalar, ScalarKind};
 
 /// Simple moving average over the last `length` values of one f64 input.
 pub struct Sma {
@@ -21,7 +21,11 @@ impl Sma {
 impl Node for Sma {
     fn schema(&self) -> NodeSchema {
         NodeSchema {
-            inputs: vec![InputSpec { kind: ScalarKind::F64, lookback: self.length }],
+            inputs: vec![InputSpec {
+                kind: ScalarKind::F64,
+                lookback: self.length,
+                firing: Firing::Any,
+            }],
             output: ScalarKind::F64,
         }
     }

crates/aura-std/src/sub.rs

@@ -3,7 +3,7 @@
 //! real fan-out + join to run (two SMAs joining into one node), exercising
 //! multi-input `Ctx` access inside a running graph.
 
-use aura_core::{Ctx, InputSpec, Node, NodeSchema, Scalar, ScalarKind};
+use aura_core::{Ctx, Firing, InputSpec, Node, NodeSchema, Scalar, ScalarKind};
 
 /// Two-input f64 difference: input 0 minus input 1. Emits `None` until both
 /// inputs have a value.
@@ -21,8 +21,8 @@ impl Node for Sub {
     fn schema(&self) -> NodeSchema {
         NodeSchema {
             inputs: vec![
-                InputSpec { kind: ScalarKind::F64, lookback: 1 },
+                InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Any },
-                InputSpec { kind: ScalarKind::F64, lookback: 1 },
+                InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Any },
             ],
             output: ScalarKind::F64,
         }