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

plan: cycle 0003 deterministic single-source sim loop

Browse Source 
Three tasks, each leaving the workspace green:
1. Sub 2-input difference node in aura-std (+ test).
2. Sim in aura-engine: Edge/Target/BootstrapError + bootstrap (schema-sizing,
   edge kind-check, Kahn topo-sort, cycle/bad-index/kind-mismatch rejection) +
   the forwarding run loop (disjoint &mut-self destructure, no per-cycle alloc),
   with 5 tests (chain, fan-out/join + determinism, cycle, kind-mismatch,
   bad-index). aura-engine gains a test-only dev-dependency on aura-std.
3. Workspace gate: build / test (26) / clippy -D warnings / purity grep.

Orchestrator decisions baked in: observe captured at eval-time (no output
column); Sim stores only fields run reads (no write-only cycle_id/source_kind
that would trip `field never read`); the sim.rs doc avoids the literal
Rc/RefCell tokens so the purity grep does not self-match.
This commit is contained in:
Brummel
2026-06-03 12:58:18 +02:00
parent aaaf463102
commit 0419e54987
1 changed files with 548 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/plans/0003-deterministic-sim-loop.md
+548
View File
@@ -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<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]))
}
}
#[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<dyn Node>,
inputs: Vec<AnyColumn>,
}
/// A bootstrapped, frozen root graph instance plus its deterministic run loop.
pub struct Sim {
nodes: Vec<NodeBox>,
topo: Vec<usize>,
out_edges: Vec<Vec<Edge>>,
source_targets: Vec<Target>,
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<Box<dyn Node>>,
source_targets: Vec<Target>,
edges: Vec<Edge>,
source_kind: ScalarKind,
observe: usize,
) -> Result<Sim, BootstrapError> {
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<NodeBox> = Vec::with_capacity(n);
for (nd, schema) in nodes.into_iter().zip(schemas.iter()) {
let inputs: Vec<AnyColumn> = 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<Edge>> = 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<usize> = (0..n).filter(|&i| indeg[i] == 0).collect();
let mut topo: Vec<usize> = 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<Item = (Timestamp, Scalar)>,
) -> Vec<Option<Scalar>> {
// 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<Item = (Timestamp, Scalar)> + '_ {
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<Box<dyn Node>> = 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<Box<dyn Node>> =
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<Box<dyn Node>> = 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<Box<dyn Node>> = 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<Box<dyn Node>> = 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<dyn Node>` is the intended
node object (set at bootstrap, dispatched by vtable), not a `dyn Any` payload, and
does not match the pattern.