diff --git a/docs/specs/0003-deterministic-sim-loop.md b/docs/specs/0003-deterministic-sim-loop.md new file mode 100644 index 0000000..1f61a42 --- /dev/null +++ b/docs/specs/0003-deterministic-sim-loop.md @@ -0,0 +1,314 @@ +# The Deterministic Single-Source Sim Loop — Design Spec + +**Date:** 2026-06-03 +**Status:** Draft — awaiting user spec review +**Authors:** orchestrator + Claude + +## Goal + +Make a **graph run**. This cycle delivers `aura-engine`'s first real content: a +deterministic, synchronous sim loop that drives one source through a wired DAG of +nodes (C8/C9), cycle by cycle in timestamp order (C4), and collects an observable +output series. It is the first point in the project where authored nodes actually +*execute* against data, not just under a hand-fed `Ctx`. + +The engine is the flat, monomorphized sharpening of RustAst's reactive push model: +where RustAst wired a pointer-graph of `Rc>` carrying boxed +`Value`s, aura uses a **flat node array + an index edge table**, topologically +ordered, owned `&mut` by the loop — no `Rc`, no `RefCell`, no `dyn Any`, no +per-cycle allocation (C1/C7). + +**In scope:** the `Sim` runtime (nodes + edges + topological schedule), its +bootstrap (size each node's input columns from its `schema`, kind-check every edge +once at wiring, reject cycles), the `run` loop (drive records, forward outputs), +and a second worked node (`Sub`) in `aura-std` so the loop is proven on a real +**fan-out + join** DAG, not a degenerate chain. + +**Deliberately out of scope** (recorded as decisions, not gaps): + +- **Freshness-gated recompute / sample-and-hold (C5).** With a single source every + node is fresh every cycle (the always-fresh degenerate case of C5). True + freshness gating is only *testable* with heterogeneous-rate inputs, which need a + second source (C3 k-way merge) or a resampler (C15). Building it now — with + nothing to exercise it — would be untestable speculative infra. It lands in + Cycle 0004 together with the second source. The loop is structured so adding it + is a gate *before* each `eval`, changing no existing type. +- **The `Source` trait and real ingestion (C3/C11).** The loop is driven by a + plain `Iterator` this cycle; the source *trait*, + data-server ingestion, and the k-way merge are the next milestone checkbox. +- **An ergonomic builder API (C19 full).** The graph is hand-constructed in the + test from nodes + edges (as Cycle 0002's node was hand-driven). The bootstrap + (size + topo + freeze) ships; the param-generic blueprint and builder ergonomics + are deferred. +- **The real sink + run registry (C18/C22).** Observation is a collected + `Vec>` for the designated output node; the sink-node abstraction + and the registry come later. +- **Parallelism across sims (C1).** One sim, single-threaded. Disjointness is a + property the design preserves; running many in parallel is a later concern. + +## Architecture + +One new module tree in `aura-engine`, plus one worked node in `aura-std`. + +A **`Sim`** is the bootstrapped, frozen root graph instance. It holds: + +- `nodes: Vec` — each a `Box` plus that node's **own** input + columns (`Vec`, sized from its `schema` at bootstrap, in slot order). + A node owning its inputs is exactly Cycle 0002's shape, so `Ctx::new(&nb.inputs)` + wraps them unchanged and no per-cycle gather/alloc is needed. +- `topo: Vec` — node indices in a valid evaluation order (Kahn's + algorithm), computed once at bootstrap. +- `out_edges: Vec>` — adjacency, indexed by producer node; each `Edge` + forwards that producer's output into a consumer's input slot. +- `source_targets: Vec` — the `(node, slot)` input slots the source value + is forwarded into each cycle. +- `cycle_id: u64` — the monotonic clock (C4). + +**Wiring is by forwarding, not shared columns.** After a producer node `n` +evaluates to `Some(v)`, the loop pushes `v` into each consumer input column its +edges name. A producer that returns `None` (filtered / not warmed up) forwards +nothing, so the consumer keeps its held window — the structural seed of +sample-and-hold, realized fully when C5 lands. Because `eval` returns an *owned* +`Option` (no borrow into the graph escapes), the read (build `Ctx`, eval) +and the write (forward the value) are cleanly separable under the borrow checker. + +**The source is the graph entry, not a node.** It has no `eval`; the loop writes +each cycle's record value directly into its target input slots. Modelling it as a +`Node` would force an `eval` with no inputs to read — the loop driving it directly +matches C4 ("one input record = one cycle") and C11 ("a source is a producer") +without contorting the node contract. + +## Concrete code shapes + +### The worked graph — what running a DAG looks like (the headline) + +This hand-wired test is the empirical evidence: a real fan-out + join DAG +(`source → SMA(2)`, `source → SMA(4)`, `(SMA2, SMA4) → Sub → observe`) runs +deterministically and produces the hand-computable series. If wiring and running a +graph is this direct, the runtime shape is right. + +(The integration test lives in `aura-engine` and uses real `aura-std` nodes, so +`aura-engine` gains a **dev-dependency** on `aura-std` — test-only; the engine +*library* still depends on `aura-core` alone and knows nothing of `aura-std`.) + +```rust +// aura-engine tests +use aura_core::{Node, Scalar, ScalarKind, Timestamp}; +use aura_engine::{Edge, Sim, Target}; +use aura_std::{Sma, Sub}; + +#[test] +fn fan_out_join_dag_runs_deterministically() { + // nodes: 0 = SMA(2), 1 = SMA(4), 2 = Sub (input 0 - input 1) + let nodes: Vec> = + vec![Box::new(Sma::new(2)), Box::new(Sma::new(4)), Box::new(Sub::new())]; + + // the source (f64 price) fans out into SMA(2).in0 and SMA(4).in0 + let source_targets = vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }]; + + // SMA outputs join into Sub: SMA(2) -> Sub.in0, SMA(4) -> Sub.in1 + let edges = vec![ + Edge { from: 0, to: 2, slot: 0 }, + Edge { from: 1, to: 2, slot: 1 }, + ]; + + // bootstrap: size input columns from schemas, kind-check edges, topo-sort. + // `observe` names the node whose per-cycle output is collected. + let mut sim = Sim::bootstrap(nodes, source_targets, edges, /*source kind*/ ScalarKind::F64, /*observe*/ 2) + .expect("valid DAG"); + + let prices = [10.0_f64, 12.0, 14.0, 16.0, 18.0, 20.0]; + let records = prices + .iter() + .enumerate() + .map(|(i, &p)| (Timestamp(i as i64), Scalar::F64(p))); + + let out = sim.run(records); + + // Sub fires only once both SMAs are warmed (SMA(4) needs 4 samples). + // At cycle 4 (0-based 3): SMA2=mean(16,14)=15, SMA4=mean(16,14,12,10)=13 -> 2. + // At cycle 5: SMA2=mean(18,16)=17, SMA4=mean(18,16,14,12)=15 -> 2. + // At cycle 6: SMA2=mean(20,18)=19, SMA4=mean(20,18,16,14)=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 yields an identical series. + let mut sim2 = Sim::bootstrap( + 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![Edge { from: 0, to: 2, slot: 0 }, Edge { from: 1, to: 2, slot: 1 }], + ScalarKind::F64, + 2, + ) + .expect("valid DAG"); + let out2 = sim2.run(prices.iter().enumerate().map(|(i, &p)| (Timestamp(i as i64), Scalar::F64(p)))); + assert_eq!(out, out2); +} +``` + +### The second worked node — `Sub` (the join) + +```rust +// aura-std/src/sub.rs +use aura_core::{Ctx, InputSpec, Node, NodeSchema, Scalar, ScalarKind}; + +/// Two-input f64 difference: input 0 minus input 1 (e.g. a fast/slow spread). +/// Emits None until both inputs have a value this cycle. +#[derive(Default)] +pub struct Sub; + +impl Sub { + 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])) + } +} +``` + +### The runtime shape (supporting, secondary) + +```rust +// aura-engine/src/sim.rs (shapes — exact bodies are the planner's job) + +/// A producer-output -> consumer-input-slot forwarding edge. +#[derive(Clone, Copy, Debug)] +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)] +pub struct Target { pub node: usize, pub slot: usize } + +struct NodeBox { + node: Box, + inputs: Vec, // this node's own input columns, sized from schema +} + +pub struct Sim { + nodes: Vec, + topo: Vec, + out_edges: Vec>, // adjacency by producer node (alloc'd at bootstrap) + source_targets: Vec, + observe: usize, // node whose per-cycle output is collected + cycle_id: u64, +} + +#[derive(Debug, PartialEq, Eq)] +pub enum BootstrapError { + /// An edge connects mismatched scalar kinds (producer output vs consumer slot). + KindMismatch { edge: Edge, producer: ScalarKind, consumer: ScalarKind }, + /// A node/slot index in an edge or target 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) — Kahn's algorithm left nodes unranked. + Cycle, +} + +impl Sim { + pub fn bootstrap( + nodes: Vec>, + source_targets: Vec, + edges: Vec, + source_kind: ScalarKind, + observe: usize, + ) -> Result { /* size inputs, kind-check, topo-sort */ } + + /// Drive the records in order; returns the observed node's per-cycle output. + pub fn run(&mut self, records: impl Iterator) + -> Vec> { /* the loop */ } +} +``` + +## Components + +| Component | Crate | Responsibility | +|-----------|-------|----------------| +| `Sim` | aura-engine | the bootstrapped frozen graph + the deterministic run loop | +| `Edge`, `Target` | aura-engine | producer→consumer and source→consumer forwarding wiring | +| `BootstrapError` | aura-engine | wiring faults caught once, at bootstrap (C7 "check paid at wiring") | +| `Sub` | aura-std | a 2-input worked node, so the loop is proven on a real fan-out+join DAG | + +## Data flow + +**Bootstrap** (once): for each node, read `schema()`, allocate its input columns +sized to each slot's `lookback`. Validate every `Edge` and `Target`: indices in +range, and producer output kind == consumer slot kind (== source kind for +targets), else `BootstrapError`. Build `out_edges` adjacency. Topologically sort +(Kahn); if any node is left unranked, the graph has a cycle → `BootstrapError::Cycle`. + +**Run** (per record): `cycle_id += 1`; forward the record's `Scalar` into each +`source_targets` slot (`push`). Then, for each node `n` in `topo` order: build +`Ctx::new(&nodes[n].inputs)`, call `eval` → owned `Option`; if `Some(v)`, +forward `v` into every consumer slot in `out_edges[n]`. After all nodes, record +`nodes[observe]`'s output for this cycle into the result vector. The forwarding +read/write split is borrow-safe because `eval` returns an owned value; per-cycle +work allocates nothing (columns are pre-sized rings; adjacency is pre-built). + +## Error handling + +- **Bootstrap faults** are values (`Result`): kind mismatch, + out-of-range index, or a wiring cycle. All are caught before any record flows — + C7's "type check paid once at wiring" generalized to the whole topology. +- **Runtime forwarding** is infallible by construction: bootstrap proved every + edge's kinds match, so the per-cycle `push` uses `.expect("kind checked at + wiring")` — a panic there means the engine is broken, never user input. +- **`None` outputs** are normal (filter / not-warmed-up), not errors: they simply + forward nothing, leaving consumers on their held window. + +## Testing strategy + +**aura-engine (`sim.rs`):** + +- *Chain*: `source → SMA(3) → observe` produces the warm-up `None`s then the + moving mean — the minimal running graph. +- *Fan-out + join + determinism* (the headline test above): `source → {SMA(2), + SMA(4)} → Sub → observe`; assert the exact series, then assert a second + identical run is bit-identical (C1). +- *Cycle rejection*: a 2-node graph wired `a → b → a` returns + `BootstrapError::Cycle`. +- *Kind mismatch*: an edge from an `f64` producer into a slot a (hypothetical) + non-`f64` consumer returns `BootstrapError::KindMismatch`. (Exercised with a + small fixture node declaring an `i64` input, or via a `Target` with a mismatched + `source_kind`.) + +**aura-std (`sub.rs`):** + +- `Sub` over two hand-fed inputs returns `a - b`, and `None` while either input is + empty. + +All four workspace gates stay green: `cargo build/test/clippy --workspace` and the +surface-purity grep (no `dyn Any` / `Rc<` / `RefCell` / per-cycle heap alloc). + +## Acceptance criteria + +1. `cargo build --workspace`, `cargo test --workspace`, + `cargo clippy --workspace --all-targets -- -D warnings` all green. +2. A hand-wired fan-out+join DAG (`source → {SMA(2), SMA(4)} → Sub → observe`) + runs and yields the exact hand-computed series — proving authored nodes execute + against data through the engine, composed as a DAG (C8/C9). +3. The same records run twice yield a bit-identical output series (C1 determinism). +4. Bootstrap rejects a wiring cycle (`BootstrapError::Cycle`) and a kind-mismatched + edge (`BootstrapError::KindMismatch`) — faults caught once, at wiring (C7). +5. The per-cycle loop allocates nothing on the node-eval path; surface purity + preserved (no new `dyn Any` / `Rc` / `RefCell`). +6. `Sub` composes in `aura-std` against the unchanged Cycle-0002 `Node`/`Ctx` + contract — no `aura-core` change is required to add a node.