diff --git a/docs/plans/0003-deterministic-sim-loop.md b/docs/plans/0003-deterministic-sim-loop.md new file mode 100644 index 0000000..95a56c1 --- /dev/null +++ b/docs/plans/0003-deterministic-sim-loop.md @@ -0,0 +1,548 @@ +# Deterministic Single-Source Sim Loop — Implementation Plan + +> **Parent spec:** `docs/specs/0003-deterministic-sim-loop.md` +> +> **For agentic workers:** REQUIRED SUB-SKILL: use the `implement` skill to run +> this plan. Steps use `- [ ]` checkboxes for tracking. + +**Goal:** Make a wired DAG of nodes run deterministically: ship `aura-engine`'s +`Sim` (bootstrap + run loop) and a 2-input `Sub` node in `aura-std`, proven on a +fan-out+join graph. + +**Architecture:** `Sub` lands first (self-contained, depends only on aura-core). +Then `Sim` in aura-engine, whose integration test uses both `Sma` (0002) and +`Sub`, so aura-engine gains a test-only dev-dependency on aura-std. Each node owns +its input columns (0002 shape); the loop forwards producer outputs into consumer +input columns; bootstrap sizes columns from schemas, kind-checks every edge, and +rejects cycles (Kahn). + +**Tech Stack:** aura-core (Node/Ctx/AnyColumn/Scalar), aura-std (nodes), aura-engine +(the Sim runtime), Rust 2024, `cargo build/test/clippy --workspace`. + +**Design decisions baked into this plan (orchestrator, from spec + recon):** +- The observed node's per-cycle output is captured **at eval time inside the loop** + (the observed node has no outgoing edge), not by reading an output column. +- `Sim` stores **only** fields `run` reads (`nodes, topo, out_edges, + source_targets, observe`). No `cycle_id`/`source_kind` field: both would be + write-only and trip `field is never read` under `-D warnings`. The C4 cycle + clock in 0003 is the record iteration itself; the explicit counter arrives in + 0004 with freshness (its first reader). `source_kind` is a bootstrap param only. +- `BootstrapError::KindMismatch { producer, consumer }` carries just the two kinds + (no synthetic edge), covering both edge and source-target mismatches. + +--- + +**Files this plan creates or modifies:** + +- Create: `crates/aura-std/src/sub.rs` — `Sub` 2-input f64-difference node + tests. +- Modify: `crates/aura-std/src/lib.rs:18-19` — add `mod sub; pub use sub::Sub;`. +- Create: `crates/aura-engine/src/sim.rs` — `Sim`, `Edge`, `Target`, `BootstrapError` + tests. +- Modify: `crates/aura-engine/src/lib.rs` — replace the doc stub; add `mod sim;` + `pub use`. +- Modify: `crates/aura-engine/Cargo.toml` — add `[dev-dependencies] aura-std`. + +--- + +### Task 1: `Sub` — the 2-input worked node + +**Files:** +- Create: `crates/aura-std/src/sub.rs` +- Modify: `crates/aura-std/src/lib.rs` + +- [ ] **Step 1: Create `crates/aura-std/src/sub.rs`** + +```rust +//! `Sub` — two-input f64 difference (input 0 minus input 1), e.g. a fast/slow +//! spread. The walking skeleton's second worked node: it gives the sim loop a +//! 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}; + +/// Two-input f64 difference: input 0 minus input 1. Emits `None` until both +/// inputs have a value. +#[derive(Default)] +pub struct Sub; + +impl Sub { + /// Build a `Sub` node. + pub fn new() -> Self { + Self + } +} + +impl Node for Sub { + fn schema(&self) -> NodeSchema { + NodeSchema { + inputs: vec![ + InputSpec { kind: ScalarKind::F64, lookback: 1 }, + InputSpec { kind: ScalarKind::F64, lookback: 1 }, + ], + 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])) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use aura_core::AnyColumn; + + #[test] + fn sub_is_difference_once_both_inputs_present() { + let mut sub = Sub::new(); + let mut inputs = vec![ + AnyColumn::with_capacity(ScalarKind::F64, 1), + AnyColumn::with_capacity(ScalarKind::F64, 1), + ]; + + // only input 0 present -> None + inputs[0].push(Scalar::F64(10.0)).unwrap(); + assert_eq!(sub.eval(Ctx::new(&inputs)), None); + + // both present -> a - b + inputs[1].push(Scalar::F64(4.0)).unwrap(); + assert_eq!(sub.eval(Ctx::new(&inputs)), Some(Scalar::F64(6.0))); + } +} +``` + +- [ ] **Step 2: Wire `sub` into `aura-std/src/lib.rs`** + +In `crates/aura-std/src/lib.rs`, replace: + +```rust +mod sma; +pub use sma::Sma; +``` + +with: + +```rust +mod sma; +mod sub; +pub use sma::Sma; +pub use sub::Sub; +``` + +- [ ] **Step 3: Verify the Sub test passes** + +Run: `cargo test -p aura-std sub_is_difference_once_both_inputs_present` +Expected: PASS (`test result: ok. 1 passed`). + +- [ ] **Step 4: Verify the crate still builds clean** + +Run: `cargo test -p aura-std` +Expected: PASS — 3 tests (2 SMA from cycle 0002 + 1 Sub). + +--- + +### Task 2: `Sim` — the deterministic run loop + +**Files:** +- Modify: `crates/aura-engine/Cargo.toml` +- Create: `crates/aura-engine/src/sim.rs` +- Modify: `crates/aura-engine/src/lib.rs` + +- [ ] **Step 1: Add the test-only dependency on aura-std** + +In `crates/aura-engine/Cargo.toml`, after the existing `[dependencies]` block +(which contains `aura-core = { path = "../aura-core" }` and two comment lines), +append: + +```toml + +[dev-dependencies] +aura-std = { path = "../aura-std" } +``` + +- [ ] **Step 2: Create `crates/aura-engine/src/sim.rs`** + +```rust +//! The deterministic single-source sim loop. A `Sim` 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). + +use aura_core::{AnyColumn, Ctx, Node, Scalar, ScalarKind, Timestamp}; + +/// A producer-output -> consumer-input-slot forwarding edge. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct Edge { + pub from: usize, + pub to: usize, + pub slot: usize, +} + +/// An input slot the source value is forwarded into each cycle. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct Target { + pub node: usize, + pub slot: usize, +} + +/// A wiring fault caught once, at bootstrap — C7's "type check paid at wiring" +/// generalized to the whole topology. +#[derive(Debug, PartialEq, Eq)] +pub enum BootstrapError { + /// An edge or source-target connects mismatched scalar kinds. + KindMismatch { producer: ScalarKind, consumer: ScalarKind }, + /// A node or slot index in an edge or target (or the observe index) is out of range. + BadIndex, + /// The wiring contains a directed cycle (only an explicit delay node may close + /// a loop; that node does not exist yet). + Cycle, +} + +struct NodeBox { + node: Box, + inputs: Vec, +} + +/// A bootstrapped, frozen root graph instance plus its deterministic run loop. +pub struct Sim { + nodes: Vec, + topo: Vec, + out_edges: Vec>, + source_targets: Vec, + observe: usize, +} + +impl Sim { + /// 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 + /// topologically orders the nodes (Kahn), rejecting any directed cycle. + pub fn bootstrap( + nodes: Vec>, + source_targets: Vec, + edges: Vec, + source_kind: ScalarKind, + 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 own input columns from its schema + 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(); + boxes.push(NodeBox { node: nd, inputs }); + } + + // source targets: the source value must match each target slot's kind + for t in &source_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 != source_kind { + return Err(BootstrapError::KindMismatch { + producer: source_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(Sim { + nodes: boxes, + topo, + out_edges, + source_targets, + observe, + }) + } + + /// Drive the records in timestamp order; returns the observed node's per-cycle + /// output. Allocates nothing on the per-cycle eval/forward path. + pub fn run( + &mut self, + records: impl Iterator, + ) -> Vec> { + // disjoint field borrows so the topo walk can read `topo`/`out_edges` + // while mutating `nodes` + let Sim { nodes, topo, out_edges, source_targets, observe } = self; + let observe = *observe; + + let mut out = Vec::new(); + for (_ts, value) in records { + // forward the source value into its target input slots + for t in source_targets.iter() { + nodes[t.node].inputs[t.slot] + .push(value) + .expect("source kind checked at wiring"); + } + + // evaluate in topological order; capture the observed output; forward + let mut observed = None; + for &nidx in topo.iter() { + 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() { + nodes[e.to].inputs[e.slot] + .push(v) + .expect("edge kind checked at wiring"); + } + } + } + out.push(observed); + } + out + } +} + +#[cfg(test)] +mod tests { + use super::*; + use aura_std::{Sma, Sub}; + + fn f64_records(prices: &[f64]) -> impl Iterator + '_ { + prices + .iter() + .enumerate() + .map(|(i, &p)| (Timestamp(i as i64), Scalar::F64(p))) + } + + #[test] + fn chain_source_sma_runs() { + // node 0 = SMA(3); source -> SMA(3).in0; observe node 0 + let nodes: Vec> = vec![Box::new(Sma::new(3))]; + let mut sim = Sim::bootstrap( + nodes, + 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])); + 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; source fans into both SMAs; SMAs join into Sub + let build = || -> Sim { + let nodes: Vec> = + vec![Box::new(Sma::new(2)), Box::new(Sma::new(4)), Box::new(Sub::new())]; + Sim::bootstrap( + nodes, + 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 = [10.0_f64, 12.0, 14.0, 16.0, 18.0, 20.0]; + + let mut sim = build(); + 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![ + 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 sim2 = build(); + let out2 = sim2.run(f64_records(&prices)); + assert_eq!(out, out2); + } + + #[test] + fn bootstrap_rejects_a_cycle() { + // two SMA(1) nodes wired a -> b -> a + let nodes: Vec> = vec![Box::new(Sma::new(1)), Box::new(Sma::new(1))]; + let err = Sim::bootstrap( + nodes, + vec![], + vec![Edge { from: 0, to: 1, slot: 0 }, Edge { from: 1, to: 0, slot: 0 }], + ScalarKind::F64, + 0, + ) + .unwrap_err(); + assert_eq!(err, BootstrapError::Cycle); + } + + #[test] + fn bootstrap_rejects_a_kind_mismatch() { + // SMA(1) declares an f64 input; feeding an i64 source mismatches + let nodes: Vec> = vec![Box::new(Sma::new(1))]; + let err = Sim::bootstrap( + nodes, + vec![Target { node: 0, slot: 0 }], + vec![], + ScalarKind::I64, + 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 nodes: Vec> = vec![Box::new(Sma::new(1))]; + let err = Sim::bootstrap( + nodes, + vec![Target { node: 0, slot: 0 }], + vec![], + ScalarKind::F64, + 5, + ) + .unwrap_err(); + assert_eq!(err, BootstrapError::BadIndex); + } +} +``` + +- [ ] **Step 3: Replace `crates/aura-engine/src/lib.rs`** + +Replace the entire current file content (a doc-only stub) with: + +```rust +//! `aura-engine` — the headless, UI-agnostic reactive SoA engine. +//! +//! Delivered in cycle 0003 — the deterministic single-source sim loop: +//! +//! - [`Sim`] — a bootstrapped, frozen root graph (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. +//! +//! Visualization is never here: it is a downstream consumer node on the streams. + +mod sim; + +pub use sim::{BootstrapError, Edge, Sim, Target}; +``` + +- [ ] **Step 4: Verify the engine builds and its tests pass** + +Run: `cargo test -p aura-engine` +Expected: PASS — 5 tests (`chain_source_sma_runs`, +`fan_out_join_dag_runs_deterministically`, `bootstrap_rejects_a_cycle`, +`bootstrap_rejects_a_kind_mismatch`, `bootstrap_rejects_a_bad_index`). + +--- + +### Task 3: Workspace gate + +**Files:** none (verification only). + +- [ ] **Step 1: Full workspace build** + +Run: `cargo build --workspace` +Expected: `Finished` — 0 errors, 0 warnings. + +- [ ] **Step 2: Full workspace test** + +Run: `cargo test --workspace` +Expected: PASS — 26 tests total, 0 failed: aura-core 18, aura-std 3 (2 SMA + 1 +Sub), aura-engine 5, aura-cli 0. No doctests (the doc comments use intra-doc +links, no executable code fences). + +- [ ] **Step 3: Clippy, warnings-as-errors** + +Run: `cargo clippy --workspace --all-targets -- -D warnings` +Expected: `Finished` — no warnings. + +- [ ] **Step 4: Surface-purity grep** + +Run: `grep -rnE 'RefCell|Rc<|dyn Any' crates/*/src` +Expected: no matches (exit code 1, no output). `Box` is the intended +node object (set at bootstrap, dispatched by vtable), not a `dyn Any` payload, and +does not match the pattern.