diff --git a/docs/specs/0005-node-output-record.md b/docs/specs/0005-node-output-record.md new file mode 100644 index 0000000..edfa682 --- /dev/null +++ b/docs/specs/0005-node-output-record.md @@ -0,0 +1,324 @@ +# Node output is a record (composite stream) — Design Spec + +**Date:** 2026-06-03 +**Status:** Draft — awaiting user spec review +**Authors:** orchestrator + Claude + +> **Parent issue:** Gitea #1 (BLOCKER, milestone *Walking skeleton*). +> Revises contract **C8**, sharpens **C7**. + +## Goal + +Generalize a node's output from a single scalar to a **record of base-scalar +columns** (K ≥ 1), with a scalar being the degenerate K = 1 case. A node still has +exactly **one output port**; that port's payload is an ordered list of named, +typed base columns (a composite stream, C7). This unblocks every multi-column +producer the engine needs next — OHLCV bars and, later, the position-event table +(C10) — none of which can exist while a node emits at most one column. + +This cycle delivers the *substrate* only: the contract change, the field-wise +binding mechanism, the Sma/Sub migration, and a neutral OHLCV proof. The position +table (C10) and broker nodes are explicitly out of scope — they are later cycles +that *consume* this capability. + +## Architecture + +Three coordinated changes, in compile-gate order: + +1. **`aura-core` — the contract.** `NodeSchema.output` changes from one + `ScalarKind` to `Vec` (an ordered list of `{ name, kind }` base + columns). `Node::eval` changes from `-> Option` to + `-> Option<&[Scalar]>`: a node fills a buffer it owns (sized once, at + construction) and returns a borrowed row of K scalars; `None` still means + *filter / not-yet-warmed*. Scalar output is the degenerate **1-field** record — + there is no separate scalar path. + +2. **`aura-std` — the degenerate migration.** `Sma` and `Sub` declare a 1-field + output (`name: "value"`), hold a `[Scalar; 1]` buffer, and return `Some(&out)`. + Behaviour is unchanged: a 1-field record is a scalar. + +3. **`aura-engine` — field-wise binding.** `Edge` gains `from_field: usize` — + which field of the producer's record this edge forwards. Bootstrap kind-checks + per field (`producer.output[from_field].kind == consumer.inputs[slot].kind`) + and range-checks `from_field`. The run loop copies a producer's returned row + into one reused scratch buffer (resolving the borrow, zero per-cycle heap + allocation once its capacity stabilizes), then scatters `scratch[from_field]` + into each out-edge's input slot. Freshness stamping is per slot, unchanged, so + the K fields of one record are **co-fresh by construction** (one `eval`, one + timestamp). + +The record is **structural bundling over the four base types (C7 intact)**, not a +fifth scalar type: no `dyn Any`, no heterogeneous buffer. Every field lands in a +base column at the edge exactly as today's single scalar does; the streamed / +stored form stays SoA. + +## Concrete code shapes + +### What the cycle delivers — the neutral OHLCV proof (test-local fixtures) + +The cycle has no end-user surface; its deliverable is the executable proof that a +node can emit K > 1 fields and a consumer can bind them individually. The +north-star slice: OHLCV as a 5-field record (the position-event table will be the +same shape), with a downstream node binding *single* fields by index. + +```rust +// A neutral multi-field producer: five f64 inputs bundled into one 5-field +// record. No trading-domain logic — it proves the output mechanism in isolation. +struct Ohlcv { + out: [Scalar; 5], +} +impl Node for Ohlcv { + 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::Barrier(0) }, + InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Barrier(0) }, + InputSpec { kind: ScalarKind::F64, lookback: 1, firing: Firing::Barrier(0) }, + ], + output: vec![ + FieldSpec { name: "open", kind: ScalarKind::F64 }, + FieldSpec { name: "high", kind: ScalarKind::F64 }, + FieldSpec { name: "low", kind: ScalarKind::F64 }, + FieldSpec { name: "close", kind: ScalarKind::F64 }, + FieldSpec { name: "volume", kind: ScalarKind::F64 }, + ], + } + } + fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> { + for i in 0..5 { + let w = ctx.f64_in(i); + if w.is_empty() { + return None; // not yet warmed + } + self.out[i] = Scalar::F64(w[0]); + } + Some(&self.out) // one 5-field record, all fields co-fresh + } +} + +// The proof wiring: five timestamp-aligned sources feed O/H/L/C/V into Ohlcv's +// barrier group (the canonical "OHLC from separate sources" case, C6), so Ohlcv +// emits one complete bar per timestamp. A downstream Sub binds field 1 (high) +// minus field 2 (low) == the bar range. Observing Sub proves field-wise binding +// routes the right columns. Nodes are [Ohlcv (0), Sub (1)]; the five sources are +// SourceSpec entries targeting Ohlcv's five input slots. +// Edge { from: 0 /*Ohlcv*/, to: 1 /*Sub*/, slot: 0, from_field: 1 /*high*/ } +// Edge { from: 0 /*Ohlcv*/, to: 1 /*Sub*/, slot: 1, from_field: 2 /*low*/ } +``` + +The **must-fail** half of the proof (wrong code must be rejected at bootstrap): + +```rust +// from_field past the producer's output width -> BadIndex +Edge { from: 0, to: 1, slot: 0, from_field: 9 } // Ohlcv has 5 fields + +// a field whose kind mismatches the consumer slot -> KindMismatch +// (e.g. binding an i64 field into an f64 input slot) +``` + +### Before → after implementation shapes (secondary) + +**`aura-core/src/node.rs`** — the contract: + +```rust +// added +pub struct FieldSpec { + pub name: &'static str, + pub kind: ScalarKind, +} + +// before +pub struct NodeSchema { + pub inputs: Vec, + pub output: ScalarKind, +} +pub trait Node { + fn schema(&self) -> NodeSchema; + fn eval(&mut self, ctx: Ctx<'_>) -> Option; +} + +// after +pub struct NodeSchema { + pub inputs: Vec, + pub output: Vec, // ordered base columns; len 1 = scalar (degenerate) +} +pub trait Node { + fn schema(&self) -> NodeSchema; + fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]>; // borrowed K-field row +} +``` + +`FieldSpec` derives `Clone, Copy, Debug, PartialEq, Eq` (like `InputSpec`); +`NodeSchema` keeps its derives. + +**`aura-std/src/sma.rs`** — the degenerate migration (Sub is analogous): + +```rust +// before +pub struct Sma { length: usize } +// schema output: ScalarKind::F64 +fn eval(&mut self, ctx: Ctx<'_>) -> Option { + let w = ctx.f64_in(0); + if w.len() < self.length { return None; } + let mut sum = 0.0; + for k in 0..self.length { sum += w[k]; } + Some(Scalar::F64(sum / self.length as f64)) +} + +// after +pub struct Sma { length: usize, out: [Scalar; 1] } +// new() initializes out: [Scalar::F64(0.0)] +// schema output: vec![FieldSpec { name: "value", kind: ScalarKind::F64 }] +fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> { + let w = ctx.f64_in(0); + if w.len() < self.length { return None; } + let mut sum = 0.0; + for k in 0..self.length { sum += w[k]; } + self.out[0] = Scalar::F64(sum / self.length as f64); + Some(&self.out) +} +``` + +**`aura-engine/src/harness.rs`** — the edge and the run loop: + +```rust +// before +pub struct Edge { pub from: usize, pub to: usize, pub slot: usize } + +// after +pub struct Edge { pub from: usize, pub to: usize, pub slot: usize, pub from_field: usize } +``` + +Bootstrap edge check — from a single `from.output` compare to a per-field one: + +```rust +// after +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 field = from.output.get(e.from_field).ok_or(BootstrapError::BadIndex)?; + let slot = to.inputs.get(e.slot).ok_or(BootstrapError::BadIndex)?; + if field.kind != slot.kind { + return Err(BootstrapError::KindMismatch { producer: field.kind, consumer: slot.kind }); + } + out_edges[e.from].push(e); +} +``` + +Run loop — the eval-and-forward, with `result: Option<&[Scalar]>` (which is +`Copy`, since `&[Scalar]` is a shared ref) and one reused `scratch` row declared +once before the loop: + +```rust +// after (per evaluated node nidx) +let result: Option<&[Scalar]> = { + let nb = &mut nodes[nidx]; + nb.node.eval(Ctx::new(&nb.inputs)) +}; +if nidx == observe { + observed = result.map(|row| row.to_vec()); +} +if let Some(row) = result { + debug_assert_eq!(row.len(), schemas_out_len[nidx]); // node honours its schema width + scratch.clear(); + scratch.extend_from_slice(row); // copy ends the producer borrow; no per-cycle alloc once warm + for e in out_edges[nidx].iter() { + let nb = &mut nodes[e.to]; + nb.inputs[e.slot].push(scratch[e.from_field]).expect("edge kind checked at wiring"); + nb.slots[e.slot] = SlotState { fresh_at: cycle_id, last_ts: ts }; + } +} +``` + +`run`'s return type changes from `Vec>` to +`Vec>>` — the observed node's full row per cycle. The +per-fire `to_vec()` is in the materialization surface (the returned collection, +already a `Vec` today), not the inter-node hot path C7's no-alloc rule governs. + +## Components + +| Component | Change | +|-----------|--------| +| `aura-core::FieldSpec` | new — `{ name: &'static str, kind: ScalarKind }`, an output column descriptor | +| `aura-core::NodeSchema` | `output: ScalarKind` → `output: Vec` | +| `aura-core::Node::eval` | `-> Option` → `-> Option<&[Scalar]>` | +| `aura-core` lib doc | update the "single output kind" line | +| `aura-std::Sma`, `::Sub` | hold an `out` buffer; 1-field schema; return `Some(&out)`; in-crate tests adjusted to the slice return | +| `aura-engine::Edge` | gains `from_field: usize` | +| `aura-engine::Harness::bootstrap` | per-field edge kind/range check | +| `aura-engine::Harness::run` | reused scratch row; field-indexed forward; `-> Vec>>` | +| `aura-engine` harness doc | the output-forward framing (one scalar → one record field) | +| `aura-engine` tests | `Ohlcv` fixture; field-binding proof; must-fail (`from_field` OOB, field kind-mismatch); existing 0003/0004 tests adapted to `from_field` + the `Vec` return | + +## Data flow + +1. The k-way merge picks the next `(timestamp, source-index)` source head and + pushes its value into target slots, stamping freshness — unchanged (C3/C4). +2. In topological order, a fired node's `eval` fills its owned buffer and returns + `Some(&row)` (or `None` to hold/filter). +3. The engine copies `row` into the reused scratch buffer (ending the producer + borrow), then for each out-edge pushes `scratch[from_field]` into the consumer + slot, stamping that slot fresh at the current `cycle_id`/`ts`. +4. Because all K fields are written in one `eval` and forwarded in one cycle, every + consumer of any field of the record sees the same timestamp — co-freshness is + structural, needing no barrier among fields (C6 stays for cross-source sync). +5. The observed node's row is cloned into the run's output vector per cycle. + +## Error handling + +- **`from_field` out of range** (`from_field >= producer.output.len()`) → + `BootstrapError::BadIndex`, caught once at wiring (C7: type check paid at + bootstrap). +- **Per-field kind mismatch** (`producer.output[from_field].kind != + consumer.inputs[slot].kind`) → `BootstrapError::KindMismatch`. +- **Node violates its declared width** (returned row length ≠ `schema.output.len()`) + → `debug_assert_eq!` in the run loop; a node-author bug, not a wiring fault, so it + is a debug assertion, not a typed `BootstrapError`. +- **Runtime per-field kind** is enforced by `AnyColumn::push` as today + (`.expect("edge kind checked at wiring")`) — the bootstrap check makes the expect + unreachable for well-typed wiring. + +## Testing strategy + +- **Degenerate compat** — the existing 0003 fan-out/join DAG and the 0004 firing + rails (AsOfSum/BarrierSum/MixedSum, diamond rejoin) keep their expected vectors; + every edge gains `from_field: 0`, every fixture returns a 1-field slice. Proves + the scalar = 1-field-record identity preserves all prior behaviour. +- **K > 1 output** — `Ohlcv` (5 f64 inputs → 5-field record); observing it yields + the full 5-field row per cycle. +- **Field-wise binding** — source → `Ohlcv` → `Sub`(high − low); observing `Sub` + proves `from_field` routes field 1 and field 2 (not field 0). A second consumer + binding different fields (e.g. close − open) on the same `Ohlcv` confirms several + edges read distinct fields of one record. +- **Co-freshness** — the field-binding consumer fires on the bar's single + timestamp (no barrier needed among the bound fields), confirming structural + co-freshness. +- **Determinism** — a bit-identical re-run of the K > 1 proof (C1). +- **Must-fail (rejection)** — `from_field` past the output width → `BadIndex`; a + field kind-mismatch → `KindMismatch`. +- **Purity** — `grep -rnE 'RefCell|Rc<|dyn Any' crates/*/src` stays empty (no + fifth type, no heterogeneous payload). + +Gates: `cargo build/test/clippy --workspace --all-targets -- -D warnings`, plus +the purity grep. + +## Acceptance criteria + +1. A node emits a record of K ≥ 1 named base columns through one output port; + `Sma`/`Sub` are the K = 1 degenerate case with unchanged behaviour. +2. An `Edge` binds exactly one producer field (`from_field`) into one consumer + slot; consuming a whole record is N such edges; there is no "bind whole record" + mechanism. +3. The K fields of one record are co-fresh by construction; C6 is untouched. +4. C7 holds: no fifth scalar type, no `dyn Any`, no heterogeneous buffer, no + per-cycle heap allocation on the inter-node forward path; the purity grep is + clean. +5. Bootstrap rejects an out-of-range `from_field` (`BadIndex`) and a per-field + kind mismatch (`KindMismatch`). +6. The neutral OHLCV proof (producer + field-binding consumer + must-fail cases) + is green; all prior 0003/0004 tests pass adapted to `from_field`. +7. The design ledger records the C8 revision and the C7 sharpening (ledger edit is + part of this cycle; the glossary `composite`/`node` record-reality pass is an + audit-time follow-up).