diff --git a/docs/plans/0004-firing-policies-and-merge.md b/docs/plans/0004-firing-policies-and-merge.md new file mode 100644 index 0000000..afd54ed --- /dev/null +++ b/docs/plans/0004-firing-policies-and-merge.md @@ -0,0 +1,1059 @@ +# Firing Policies + Ingestion Merge — Implementation Plan + +> **Parent spec:** `docs/specs/0004-firing-policies-and-merge.md` +> +> **For agentic workers:** REQUIRED SUB-SKILL: use the `implement` skill to run +> this plan. Steps use `- [ ]` checkboxes for tracking. + +**Goal:** Ship both firing policies (C6 — A fire-on-any-fresh+hold, B all-fresh +barrier) on top of a k-way ingestion merge (C3), the monotonic cycle_id clock +(C4), and freshness-gated recompute (C5), fully tested on both rails. + +**Architecture:** `Firing` is a per-input enum on `InputSpec` (aura-core). The +engine merges N in-memory timestamped sources into one chronological cycle stream +(linear-scan k-way, ties by source index), stamping each input slot per push with +two read clocks — `fresh_at` (cycle_id, freshness epoch C5) and `last_ts` +(timestamp, barrier token C6). A node fires when any `Any` input is fresh or any +`Barrier` group has all members at the current timestamp; a non-firing node holds +(no push → consumers see the held value via `window[0]`). + +**Tech Stack:** aura-core (`Firing`, `InputSpec`), aura-std (`Sma`/`Sub` firing +annotation), aura-engine (`SourceSpec`, `SlotState`, `Harness::bootstrap`/`run`, +the `fires` predicate). Gates: `cargo build/test/clippy --workspace -- -D +warnings`, surface-purity grep. + +**Note on validation gates:** every inlined code body below is Rust (the project's +source language), so the per-task `cargo build`/`cargo test` gates are its +validation — there is no separate surface-language parse gate (the profile +declares no `spec_validation`; the parse-every-block gate is a documented no-op). + +**Task ordering rationale (compile-gate discipline):** Task 1 adds a field to +`InputSpec`, which breaks `Sma`/`Sub` — both callers are threaded inside Task 1, +so it ends on a full green workspace. Task 2 changes the `bootstrap`/`run` +signatures, which breaks the in-module tests; Task 2 therefore ends on a +**partial** gate (`cargo build -p aura-engine --lib`, which excludes `#[cfg(test)]`), +and Task 3 finishes threading by rewriting the tests, ending on the full +workspace gate. + +--- + +## Files this plan creates or modifies + +- Modify: `crates/aura-core/src/node.rs` — add `Firing` enum + `firing` field on `InputSpec`; fix module doc. +- Modify: `crates/aura-core/src/lib.rs:41` — re-export `Firing`; update roadmap doc. +- Modify: `crates/aura-std/src/sma.rs:6,24` — import `Firing`; declare `Firing::Any`. +- Modify: `crates/aura-std/src/sub.rs:6,23-25` — import `Firing`; declare `Firing::Any` (×2). +- Modify: `crates/aura-engine/src/harness.rs` — `SourceSpec`, `SlotState`, `NodeBox`/`Harness` fields, `Debug`, `bootstrap`, `run`, `fires`/`group_id`, module doc; rewrite the test module. +- Modify: `crates/aura-engine/src/lib.rs:24` — re-export `SourceSpec`; update module doc. +- Test: `crates/aura-core/src/node.rs` — `input_spec_carries_firing`. +- Test: `crates/aura-engine/src/harness.rs` — 8 tests (chain, fan-out compat+determinism, mode A, mode B, mixed A+B, three bootstrap rejects). + +--- + +## Task 1: aura-core `Firing` + `InputSpec.firing`, threaded through aura-std + +**Files:** +- Modify: `crates/aura-core/src/node.rs` +- Modify: `crates/aura-core/src/lib.rs:41` +- Modify: `crates/aura-std/src/sma.rs:6,24` +- Modify: `crates/aura-std/src/sub.rs:6,23-25` + +- [ ] **Step 1: Add the `Firing` enum and the `firing` field, fix the module doc** + +In `crates/aura-core/src/node.rs`, replace the module-doc lines 1-4: + +```rust +//! 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 +//! `eval`. Firing policy (C6) and tunable params (C12/C19) are deliberately not +//! part of the schema yet — see spec 0002's "Out of scope". +``` + +with: + +```rust +//! The node contract (C8): the interface every node implements. A node declares +//! its inputs (each with its scalar kind, lookback depth, and firing policy, C6) +//! and its output kind via `schema`, and computes one cycle's output via `eval`. +//! Tunable params (C12/C19) are deliberately not part of the schema yet — see +//! spec 0002's "Out of scope". +``` + +Then replace the `InputSpec` definition (lines 8-14): + +```rust +/// One declared input of a node: its scalar kind and the lookback depth the +/// engine must pre-size for it (must be >= 1). +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct InputSpec { + pub kind: ScalarKind, + pub lookback: usize, +} +``` + +with the new `Firing` enum followed by the extended `InputSpec`: + +```rust +/// The firing policy of one input (C6): how the engine decides whether a node +/// 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, +} +``` + +- [ ] **Step 2: Add the `input_spec_carries_firing` test to node.rs** + +Append to `crates/aura-core/src/node.rs` (the file currently ends after the +`Node` trait at line 31, with no test module): + +```rust + +#[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); + } +} +``` + +- [ ] **Step 3: Re-export `Firing` and update the aura-core roadmap doc** + +In `crates/aura-core/src/lib.rs`, replace line 41: + +```rust +pub use node::{InputSpec, Node, NodeSchema}; +``` + +with: + +```rust +pub use node::{Firing, InputSpec, Node, NodeSchema}; +``` + +Then replace the roadmap doc (lines 26-28): + +```rust +//! Still to come (subsequent cycles): the firing policies (A: fire-on-any-fresh + +//! hold; B: all-fresh barrier), the deterministic sim loop, sources, and +//! ingestion. +``` + +with: + +```rust +//! Still to come (subsequent cycles): the `Source` trait + data-server ingestion +//! and source-native time normalization (C3/C11), the broker-independent +//! position-event output and downstream broker nodes (C10), and the run registry +//! (C18/C22). +``` + +- [ ] **Step 4: Thread `Firing::Any` through `Sma`** + +In `crates/aura-std/src/sma.rs`, replace the import line 6: + +```rust +use aura_core::{Ctx, InputSpec, Node, NodeSchema, Scalar, ScalarKind}; +``` + +with: + +```rust +use aura_core::{Ctx, Firing, InputSpec, Node, NodeSchema, Scalar, ScalarKind}; +``` + +Then replace the `schema` inputs line 24: + +```rust + inputs: vec![InputSpec { kind: ScalarKind::F64, lookback: self.length }], +``` + +with: + +```rust + inputs: vec![InputSpec { + kind: ScalarKind::F64, + lookback: self.length, + firing: Firing::Any, + }], +``` + +- [ ] **Step 5: Thread `Firing::Any` through `Sub`** + +In `crates/aura-std/src/sub.rs`, replace the import line 6: + +```rust +use aura_core::{Ctx, InputSpec, Node, NodeSchema, Scalar, ScalarKind}; +``` + +with: + +```rust +use aura_core::{Ctx, Firing, InputSpec, Node, NodeSchema, Scalar, ScalarKind}; +``` + +Then replace the two `InputSpec` constructions (lines 23-26): + +```rust + inputs: vec![ + InputSpec { kind: ScalarKind::F64, lookback: 1 }, + InputSpec { kind: ScalarKind::F64, lookback: 1 }, + ], +``` + +with: + +```rust + inputs: vec![ + InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Any }, + InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Any }, + ], +``` + +- [ ] **Step 6: Build and test the full workspace** + +Run: `cargo build --workspace` +Expected: 0 errors (aura-engine still uses the old single-source API, untouched +by this task — it reads `schema.inputs[i].kind`/`.lookback`, not constructing +`InputSpec`, so the new field does not break it). + +Run: `cargo test --workspace` +Expected: PASS. aura-core gains `input_spec_carries_firing` (19 passed); aura-std +`Sma`/`Sub` tests unchanged (3 passed); aura-engine's 5 existing tests still pass +(5 passed). 27 total, 0 failed. + +Run: `cargo clippy --workspace --all-targets -- -D warnings` +Expected: 0 warnings. + +--- + +## Task 2: aura-engine production — `SourceSpec`, merge, firing gate + +**Files:** +- Modify: `crates/aura-engine/src/harness.rs` (production code only; the test module is rewritten in Task 3) +- Modify: `crates/aura-engine/src/lib.rs:24` + +- [ ] **Step 1: Rewrite the harness.rs module doc + import `Firing`** + +In `crates/aura-engine/src/harness.rs`, replace the module-doc block (lines 1-11): + +```rust +//! The harness — the closed root graph that runs (glossary: `harness`, C20) — +//! and its deterministic single-source run loop. A `Harness` is a bootstrapped, +//! frozen root graph: a flat node array plus an index edge table, topologically +//! ordered, driven cycle by cycle by one source. It is the flat, monomorphized +//! sharpening of RustAst's reference-counted, interior-mutable observer push +//! graph: no reference counting, no interior mutability, no per-cycle allocation +//! (C1/C7). Each node owns its input columns (the cycle-0002 shape), so `Ctx` is +//! unchanged; a producer's `eval` +//! 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 +//! later cycle). +``` + +with: + +```rust +//! The harness — the closed root graph that runs (glossary: `harness`, C20) — +//! and its deterministic run loop. A `Harness` is a bootstrapped, frozen root +//! graph: a flat node array plus an index edge table, topologically ordered, +//! driven cycle by cycle by a k-way merge of timestamped sources (C3/C4). It is +//! the flat, monomorphized sharpening of RustAst's reference-counted, +//! interior-mutable observer push graph: no reference counting, no interior +//! mutability, no per-cycle allocation (C1/C7). Each node owns its input columns +//! (the cycle-0002 shape), so `Ctx` is unchanged. +//! +//! Firing (C5/C6) gates re-evaluation. Two read clocks drive it, both stamped per +//! 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. +``` + +Then replace the import line 13: + +```rust +use aura_core::{AnyColumn, Ctx, Node, Scalar, ScalarKind, Timestamp}; +``` + +with: + +```rust +use aura_core::{AnyColumn, Ctx, Firing, Node, Scalar, ScalarKind, Timestamp}; +``` + +- [ ] **Step 2: Add `SourceSpec` after the `Target` struct** + +In `crates/aura-engine/src/harness.rs`, immediately after the `Target` struct +(which ends at line 28 with its closing `}`), insert: + +```rust + +/// 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, +} +``` + +- [ ] **Step 3: Add `SlotState` and extend `NodeBox`** + +In `crates/aura-engine/src/harness.rs`, replace the `NodeBox` struct (lines 43-46): + +```rust +struct NodeBox { + node: Box, + inputs: Vec, +} +``` + +with the `SlotState` declaration followed by the extended `NodeBox`: + +```rust +/// 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, + inputs: Vec, + firing: Vec, + slots: Vec, +} +``` + +- [ ] **Step 4: Rename the `Harness.source_targets` field and its `Debug` print** + +In `crates/aura-engine/src/harness.rs`, replace the `Harness` struct (lines 48-55): + +```rust +pub struct Harness { + nodes: Vec, + topo: Vec, + out_edges: Vec>, + source_targets: Vec, + observe: usize, +} +``` + +with: + +```rust +pub struct Harness { + nodes: Vec, + topo: Vec, + out_edges: Vec>, + sources: Vec, + observe: usize, +} +``` + +Then in the manual `Debug` impl, replace the `source_targets` field-print line 67: + +```rust + .field("source_targets", &self.source_targets) +``` + +with: + +```rust + .field("sources", &self.sources) +``` + +- [ ] **Step 5: Rewrite `bootstrap` to take `sources: Vec`** + +In `crates/aura-engine/src/harness.rs`, replace the entire `bootstrap` function +(lines 73-159, from ` /// Bind nodes + wiring` through the closing ` }` of +`bootstrap`) with: + +```rust + /// Bind nodes + wiring into a frozen, runnable graph. Sizes each node's input + /// 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>, + sources: Vec, + edges: Vec, + observe: usize, + ) -> Result { + let n = nodes.len(); + if observe >= n { + return Err(BootstrapError::BadIndex); + } + + let schemas: Vec<_> = nodes.iter().map(|nd| nd.schema()).collect(); + + // size each node's input columns from its schema; lift firing; init slots + let mut boxes: Vec = Vec::with_capacity(n); + for (nd, schema) in nodes.into_iter().zip(schemas.iter()) { + let inputs: Vec = schema + .inputs + .iter() + .map(|spec| AnyColumn::with_capacity(spec.kind, spec.lookback)) + .collect(); + let firing: Vec = schema.inputs.iter().map(|spec| spec.firing).collect(); + let slots: Vec = schema + .inputs + .iter() + .map(|_| SlotState { fresh_at: 0, last_ts: Timestamp(i64::MIN) }) + .collect(); + boxes.push(NodeBox { node: nd, inputs, firing, slots }); + } + + // source targets: each source's value must match each of its target slots' kind + for src in &sources { + for t in &src.targets { + let s = schemas.get(t.node).ok_or(BootstrapError::BadIndex)?; + let slot = s.inputs.get(t.slot).ok_or(BootstrapError::BadIndex)?; + if slot.kind != src.kind { + return Err(BootstrapError::KindMismatch { + producer: src.kind, + consumer: slot.kind, + }); + } + } + } + + // edges: indices in range, producer output kind == consumer slot kind + let mut out_edges: Vec> = vec![Vec::new(); n]; + for &e in &edges { + let from = schemas.get(e.from).ok_or(BootstrapError::BadIndex)?; + let to = schemas.get(e.to).ok_or(BootstrapError::BadIndex)?; + let slot = to.inputs.get(e.slot).ok_or(BootstrapError::BadIndex)?; + if from.output != slot.kind { + return Err(BootstrapError::KindMismatch { + producer: from.output, + consumer: slot.kind, + }); + } + out_edges[e.from].push(e); + } + + // Kahn topological sort; a leftover node means a cycle + let mut indeg = vec![0usize; n]; + for &e in &edges { + indeg[e.to] += 1; + } + let mut queue: Vec = (0..n).filter(|&i| indeg[i] == 0).collect(); + let mut topo: Vec = Vec::with_capacity(n); + let mut head = 0; + while head < queue.len() { + let u = queue[head]; + head += 1; + topo.push(u); + for e in &out_edges[u] { + indeg[e.to] -= 1; + if indeg[e.to] == 0 { + queue.push(e.to); + } + } + } + if topo.len() != n { + return Err(BootstrapError::Cycle); + } + + Ok(Harness { + nodes: boxes, + topo, + out_edges, + sources, + observe, + }) + } +``` + +- [ ] **Step 6: Rewrite `run` as the merge + firing loop** + +In `crates/aura-engine/src/harness.rs`, replace the entire `run` function +(lines 161-202, from ` /// Drive the records` through the closing ` }` of +`run`) with: + +```rust + /// Drive the sources, k-way-merged in timestamp order (ties by source index, + /// C4); returns the observed node's per-cycle emission (`Some` when it fired + /// and produced output, `None` when it held or filtered). One stream per + /// source, each ascending in timestamp (C3 ingestion precondition). Allocates + /// nothing per cycle beyond the output vector. + pub fn run(&mut self, streams: Vec>) -> Vec> { + assert_eq!( + streams.len(), + 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 = vec![0; streams.len()]; + let mut cycle_id: u64 = 0; + let mut out = Vec::new(); + + loop { + // pick the live source head with the smallest (timestamp, source index) + let mut pick: Option = None; + for (s, stream) in streams.iter().enumerate() { + 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; 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)) + }; + if nidx == observe { + observed = result; + } + if let Some(v) = result { + for e in out_edges[nidx].iter() { + let nb = &mut nodes[e.to]; + nb.inputs[e.slot].push(v).expect("edge kind checked at wiring"); + nb.slots[e.slot] = SlotState { fresh_at: cycle_id, last_ts: ts }; + } + } + } + out.push(observed); + } + out + } +``` + +- [ ] **Step 7: Add the `fires` predicate and `group_id` helper** + +In `crates/aura-engine/src/harness.rs`, immediately after the closing `}` of the +`impl Harness { ... }` block (after `run`), insert the two free functions: + +```rust + +/// 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 { + match f { + Firing::Any => None, + Firing::Barrier(g) => Some(*g), + } +} +``` + +- [ ] **Step 8: Re-export `SourceSpec` and update the aura-engine module doc** + +In `crates/aura-engine/src/lib.rs`, replace the module-doc lines 3-18: + +```rust +//! Delivered in cycle 0003 — the harness and its deterministic single-source +//! run loop: +//! +//! - [`Harness`] — the closed root graph that runs (a flat node array + an index +//! edge table, topologically ordered) and its deterministic `run` loop: one +//! source driven through a wired DAG of nodes, cycle by cycle, each output +//! forwarded into its consumers' input columns (C1/C4/C7/C8/C9). +//! - [`Edge`] / [`Target`] — producer->consumer and source->consumer wiring. +//! - [`BootstrapError`] — wiring faults caught once, at bootstrap (kind +//! mismatch, bad index, directed cycle). +//! +//! Still to come (subsequent cycles): freshness-gated recompute / sample-and-hold +//! (C5), the ingestion boundary (k-way merge of timestamped sources, C3), the +//! broker-independent position-event output and downstream broker nodes (C10), +//! and the atomic sim unit `(topology + params + data-window + seed) -> metrics` +//! that the sweep / optimize / walk-forward / Monte-Carlo axes orchestrate. +``` + +with: + +```rust +//! Delivered across cycles 0003-0004 — the harness and its deterministic run +//! loop: +//! +//! - [`Harness`] — the closed root graph that runs (a flat node array + an index +//! edge table, topologically ordered) and its deterministic `run` loop: a +//! k-way merge of timestamped sources (C3/C4) driven through a wired DAG of +//! nodes, cycle by cycle, with freshness-gated recompute and the two firing +//! 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): the `Source` trait + data-server ingestion +//! and source-native time normalization (C3/C11), the broker-independent +//! position-event output and downstream broker nodes (C10), and the atomic sim +//! unit `(topology + params + data-window + seed) -> metrics` that the sweep / +//! optimize / walk-forward / Monte-Carlo axes orchestrate. +``` + +Then replace the re-export line 24: + +```rust +pub use harness::{BootstrapError, Edge, Harness, Target}; +``` + +with: + +```rust +pub use harness::{BootstrapError, Edge, Harness, SourceSpec, Target}; +``` + +- [ ] **Step 9: Partial build gate (library only; the test module is rewritten in Task 3)** + +Run: `cargo build -p aura-engine --lib` +Expected: 0 errors. This compiles the aura-engine library without its +`#[cfg(test)]` module (which still references the old `bootstrap`/`run` signatures +and is rewritten in Task 3). Do **not** run `cargo test` or +`cargo build --all-targets` in this task — the test module does not compile yet, +by design. + +Run: `cargo clippy -p aura-engine --lib -- -D warnings` +Expected: 0 warnings. + +--- + +## Task 3: aura-engine tests — rewrite the 5 existing onto the new API, add the firing rails + +**Files:** +- Modify: `crates/aura-engine/src/harness.rs` (the `#[cfg(test)] mod tests` block, lines 205-328) + +- [ ] **Step 1: Replace the entire test module** + +In `crates/aura-engine/src/harness.rs`, replace the whole `#[cfg(test)] mod tests` +block (lines 205-328, from `#[cfg(test)]` to the final closing `}` of the module) +with: + +```rust +#[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}; + + /// Build an f64 source stream from (timestamp, value) points. + fn f64_stream(points: &[(i64, f64)]) -> Vec<(Timestamp, Scalar)> { + points.iter().map(|&(t, v)| (Timestamp(t), Scalar::F64(v))).collect() + } + + // --- 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 { + 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 { + 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 { + 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); one source -> SMA(3).in0; observe node 0 + let mut h = Harness::bootstrap( + vec![Box::new(Sma::new(3))], + vec![SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }], + vec![], + 0, + ) + .expect("valid"); + 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![ + None, + None, + Some(Scalar::F64(2.0)), + Some(Scalar::F64(3.0)), + Some(Scalar::F64(4.0)), + ] + ); + } + + #[test] + fn fan_out_join_dag_runs_deterministically() { + // 0 = SMA(2), 1 = SMA(4), 2 = Sub; one source fans into both SMAs (all + // Any), SMAs join into Sub — the 0003 compat baseline on the new API. + let build = || { + Harness::bootstrap( + vec![Box::new(Sma::new(2)), Box::new(Sma::new(4)), Box::new(Sub::new())], + 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 }], + 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 mut h = build(); + let out = h.run(vec![prices.clone()]); + // Sub fires once SMA(4) is warm (cycle 4): 15-13, 17-15, 19-17 -> 2. + assert_eq!( + out, + vec![ + None, + None, + None, + Some(Scalar::F64(2.0)), + Some(Scalar::F64(2.0)), + Some(Scalar::F64(2.0)), + ] + ); + + // determinism (C1): a second identical run is bit-identical + let mut h2 = build(); + assert_eq!(h2.run(vec![prices]), out); + } + + #[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 err = Harness::bootstrap( + 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 }], + 0, + ) + .unwrap_err(); + assert_eq!(err, BootstrapError::Cycle); + } + + #[test] + fn bootstrap_rejects_a_kind_mismatch() { + // SMA(1) declares an f64 input; an i64 source mismatches + let err = Harness::bootstrap( + vec![Box::new(Sma::new(1))], + vec![SourceSpec { kind: ScalarKind::I64, targets: vec![Target { node: 0, slot: 0 }] }], + vec![], + 0, + ) + .unwrap_err(); + assert_eq!( + err, + BootstrapError::KindMismatch { producer: ScalarKind::I64, consumer: ScalarKind::F64 } + ); + } + + #[test] + fn bootstrap_rejects_a_bad_index() { + // observe node 5 does not exist + let err = Harness::bootstrap( + vec![Box::new(Sma::new(1))], + vec![SourceSpec { kind: ScalarKind::F64, targets: vec![Target { node: 0, slot: 0 }] }], + vec![], + 5, + ) + .unwrap_err(); + assert_eq!(err, BootstrapError::BadIndex); + } +} +``` + +- [ ] **Step 2: Full workspace build** + +Run: `cargo build --workspace --all-targets` +Expected: 0 errors (the test module now compiles against the new API). + +- [ ] **Step 3: Full workspace test suite** + +Run: `cargo test --workspace` +Expected: PASS, 0 failed. Per-crate: aura-core 19, aura-std 3, aura-engine 8 +(30 total). The 8 aura-engine tests are: `chain_source_sma_runs`, +`fan_out_join_dag_runs_deterministically`, +`mode_a_as_of_fires_on_any_fresh_and_holds`, +`mode_b_barrier_fires_only_on_timestamp_coincidence`, +`mixed_a_and_b_or_combine_on_one_node`, `bootstrap_rejects_a_cycle`, +`bootstrap_rejects_a_kind_mismatch`, `bootstrap_rejects_a_bad_index`. + +- [ ] **Step 4: Lint gate** + +Run: `cargo clippy --workspace --all-targets -- -D warnings` +Expected: 0 warnings. + +- [ ] **Step 5: Surface-purity grep** + +Run: `grep -rnE 'RefCell|Rc<|dyn Any' crates/*/src` +Expected: no output (exit code 1) — the harness module doc phrases RustAst's model +as "reference-counted, interior-mutable" without the literal `Rc<`/`RefCell`/`dyn +Any` tokens, and no production code uses them.