diff --git a/docs/specs/0006-sink-recording.md b/docs/specs/0006-sink-recording.md new file mode 100644 index 0000000..5808b70 --- /dev/null +++ b/docs/specs/0006-sink-recording.md @@ -0,0 +1,330 @@ +# Sink recording — recording is a node role, not a type — Design Spec + +**Date:** 2026-06-04 +**Status:** Draft — awaiting user spec review +**Authors:** orchestrator + Claude + +## Goal + +Replace the engine's single `observe: usize` recording affordance with real +recording-by-node, so that one run can record *many* streams instead of exactly +one node's row. This is the BLOCKER (#2) that the milestone's multi-producer × +multi-consumer × multi-sink stress matrix (#3) structurally depends on. + +The cycle settles a conceptual point and makes the substrate reflect it: **a +"sink" is a role, not a type.** Recording is a node, in its `eval`, pushing a +value to a destination it holds as a field (a channel, a chart handle, a buffer) +— an out-of-graph side effect. There is no `Sink` type, no engine sink concept, +no structural marker. A node that only records is the pure-consumer case (C8); a +node may *also* return an output the engine forwards downstream, recording and +producing in the same `eval` (C8 "producer, consumer, or both"). + +The load-bearing principle, recorded in the ledger by this cycle: **in-graph +routing stays engine-owned data (the edge table — introspectable, topologically +ordered, kind-checked at wiring); the escape *out* of the graph is a node-side +side effect. The boundary between them is exactly the determinism / +graph-as-data boundary** (C1/C7/C9, and the recording boundary of C18/C22). + +Closes #2. Unblocks #3. No trading domain (no position table, no broker, no +equity) — pure compute substrate. + +## Architecture + +Three moving parts, all small, because the substrate already carries a sink as +an ordinary node: + +1. **`Ctx::now()` (aura-core).** A recording node stamps each record with the + cycle's timestamp. The cycle timestamp is the engine's clock, not any node's + output, so it must reach `eval` through `Ctx`. `Ctx` gains a `now: Timestamp` + field and a `now()` accessor. Reading "now" is causal — it is the present + cycle's timestamp, never the future (C2) — so it introduces no look-ahead. + `Ctx` stays `Copy` (Timestamp is `Copy`). + +2. **`observe` is removed (aura-engine).** The `observe: usize` field on + `Harness`, its `observe >= n` bootstrap `BadIndex` check, and the per-cycle + "collect the observed node's row" logic all go. `Harness::bootstrap` drops its + fourth parameter. `Harness::run` no longer yields a stream — it returns `()`. + Recording is now a node-side concern; retention is owned by whoever + constructed the recording node and holds its destination's read end. + +3. **The run loop is otherwise unchanged as a router.** It still evaluates nodes + in topological order, gates by firing (C5/C6), and scatters each fired + producer's returned row field-wise along `out_edges` into consumer slots + (C7/0005). A recording node's `eval` reads its typed input windows + `now()`, + pushes a record to its held destination, and returns `None` (pure sink) or + `Some(row)` (producer-and-sink). The engine forwards the `Some` arm exactly as + today and is wholly oblivious to the side-effect push. The one change inside + the loop is that `Ctx` is now constructed with the cycle `ts`. + +What the engine does **not** gain: a `Sink` type, a `SinkHandler` trait, any +`dyn FnMut`, any engine registry of recording points, any "is this a sink" flag. +A recording node is identified by nothing structural — the constructing World +knows its recording points because it built them. + +## Concrete code shapes + +### User-facing program (the worked example, = the acceptance evidence) + +Recording two distinct interior streams of one run — the SMA(2) and SMA(4) +outputs of a fan-out DAG — into two independent recording nodes. The destination +shown is a channel, the faithful miniature of a real recording sink (a chart on +another thread, the run registry): the node holds the `Sender`, the caller holds +the `Receiver` and owns retention. + +`Recorder` here is **this cycle's test-local fixture**, not a shipped type — it +stands in for the recording node a downstream author writes (their `ChartSink`, +their `RegistrySink`), each a node with `output: vec![]` of the same shape. +Nothing in the engine surface is a `Sink`; the example shows the *shape* every +recording node has. + +```rust +use std::sync::mpsc; +use aura_core::{Scalar, ScalarKind, Timestamp}; +use aura_engine::{Edge, Harness, SourceSpec, Target}; +use aura_std::Sma; + +// Two recording nodes, each tapping one indicator. `Recorder` is a node with +// `output: vec![]` that, on every fired cycle, reads its one f64 input and sends +// `(now, value)` to a channel the caller drains. +let (tx_fast, rx_fast) = mpsc::channel(); +let (tx_slow, rx_slow) = mpsc::channel(); + +let mut h = Harness::bootstrap( + vec![ + Box::new(Sma::new(2)), // 0 + Box::new(Sma::new(4)), // 1 + Box::new(Recorder::new(&[ScalarKind::F64], tx_fast)), // 2: taps node 0 + Box::new(Recorder::new(&[ScalarKind::F64], tx_slow)), // 3: taps node 1 + ], + 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, from_field: 0 }, // SMA(2) -> recorder fast + Edge { from: 1, to: 3, slot: 0, from_field: 0 }, // SMA(4) -> recorder slow + ], + // no observe argument +) +.expect("valid DAG"); + +h.run(vec![f64_stream(&[(1, 10.0), (2, 12.0), (3, 14.0), (4, 16.0), (5, 18.0)])]); + +let fast: Vec<(Timestamp, Vec)> = rx_fast.try_iter().collect(); +let slow: Vec<(Timestamp, Vec)> = rx_slow.try_iter().collect(); +// `fast` holds SMA(2)'s stream, `slow` holds SMA(4)'s — two streams, one run. +``` + +### Producer-and-sink in one node (the C8 "both" case) + +A node may record *and* return an output the engine forwards. This is not a +special node kind — just a normal `eval` with a side effect before its return: + +```rust +fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Scalar]> { + let v = ctx.f64_in(0)[0]; + let _ = self.tx.send((ctx.now(), vec![Scalar::F64(v)])); // sink side effect: out of graph + self.out[0] = Scalar::F64(v); + Some(&self.out) // producer output: engine forwards it +} +``` + +### Must-fail fixture (correct behaviour is rejection) + +A recording node declares typed input slots like any consumer, so an edge whose +producer field kind does not match the slot kind is rejected at bootstrap — the +existing per-field kind check (0005) already covers it; recording adds no new +hole: + +```rust +// Recorder declares an f64 input; an i64 producer field bound into it is rejected. +let err = Harness::bootstrap( + vec![Box::new(I64Source { /* one i64 output field */ }), + Box::new(Recorder::new(&[ScalarKind::F64], tx))], + vec![/* ... */], + vec![Edge { from: 0, to: 1, slot: 0, from_field: 0 }], // i64 field -> f64 slot + // no observe +) +.unwrap_err(); +assert_eq!(err, BootstrapError::KindMismatch { + producer: ScalarKind::I64, consumer: ScalarKind::F64, +}); +``` + +### Implementation shape (before → after — supporting, not the headline) + +`Ctx` (aura-core/src/ctx.rs): + +```rust +// before +pub struct Ctx<'a> { inputs: &'a [AnyColumn] } +impl<'a> Ctx<'a> { + pub fn new(inputs: &'a [AnyColumn]) -> Self { Self { inputs } } +} + +// after +pub struct Ctx<'a> { inputs: &'a [AnyColumn], now: Timestamp } +impl<'a> Ctx<'a> { + pub fn new(inputs: &'a [AnyColumn], now: Timestamp) -> Self { Self { inputs, now } } + /// The current cycle's timestamp (C4). Causal — the present, never the + /// future (C2). + pub fn now(&self) -> Timestamp { self.now } +} +``` + +`Harness` (aura-engine/src/harness.rs): + +```rust +// before +pub struct Harness { nodes, topo, out_edges, sources, observe: usize } +pub fn bootstrap(nodes, sources, edges, observe: usize) -> Result +pub fn run(&mut self, streams) -> Vec>> + +// after +pub struct Harness { nodes, topo, out_edges, sources } // no observe +pub fn bootstrap(nodes, sources, edges) -> Result // no observe param +pub fn run(&mut self, streams) // returns () +``` + +Inside `run` the only change is the `Ctx` construction and the deletion of the +observe bookkeeping: + +```rust +// before: Ctx::new(&nb.inputs); ... if nidx == observe { observed = result.map(...) } ... out.push(observed) +// after: Ctx::new(&nb.inputs, ts); (no observe branch, no out vector, no return) +``` + +## Components + +- **aura-core** — `Ctx` gains `now: Timestamp` + `now()`; `Ctx::new` takes the + timestamp. The three in-crate `Ctx::new` call sites (its unit tests) pass a + timestamp. `Node`/`NodeSchema`/`FieldSpec`/`Scalar` are untouched — a recording + node uses the existing `eval -> Option<&[Scalar]>` contract verbatim. +- **aura-engine** — `Harness` loses `observe` (field, bootstrap check, per-cycle + collection); `bootstrap` loses its fourth parameter; `run` returns `()` and + constructs `Ctx` with `ts`. A test-local `Recorder` fixture (see Testing + strategy) plus the migrated and new tests. No new public type ships — the + engine surface *shrinks*. +- **design ledger** — Realization (cycle 0006) notes on **C8** (the pure-consumer + / sink half is now realized: recording = an `eval` side effect to a held + destination; "sink" is a role, not a type; a node may produce and record at + once) and **C22** (sinks-as-recording-mechanism is realized at the substrate + level; the recorded trace is what a recording node pushed out; the in-graph + routing vs out-of-graph escape boundary = the determinism boundary). No new + contract; the `Harness` API change (observe removed) is recorded. + +## Data flow + +Per source tick: the engine advances one cycle with timestamp `ts` (C4), +forwards the source value into its target slots, then evaluates nodes in +topological order. A fired producer returns a row; the engine copies it into the +reused scratch buffer and scatters `scratch[from_field]` into each out-edge's +consumer slot (unchanged, C7/0005). A fired recording node reads its typed input +windows (newest of each, `ctx._in(i)[0]`) plus `ctx.now()`, assembles a +record, and pushes it to the destination it holds; it returns `None` (pure sink) +or `Some(row)` (also a producer). Nothing the recording node pushes re-enters +the graph — a recording node has no out-edges for its recorded values (a "both" +node's *forwarded* output is a separate, declared port). + +Recorded streams are **sparse and timestamped**: a recording node emits a record +only on the cycles it fires (its firing policy — Any or Barrier — sets the rate), +each tagged with `now()`. This supersedes the old dense `Vec>` +(one entry per cycle) and matches a trace of timestamped events (C18/C22); +streams of different-rate recorders no longer have to share a cycle index. + +## Error handling + +- A recording node's input slots are kind-checked at bootstrap exactly like any + consumer's (the per-field `KindMismatch` and range `BadIndex` from 0005). A + mis-wired recording edge fails at bootstrap, before any data flows. +- The `observe >= n` `BadIndex` check is removed with `observe`. No + `BootstrapError` variant is added or removed otherwise. +- `ctx.now()` is total — every cycle has a timestamp; there is no cold/None case. +- The test destination is an `mpsc::Sender`; a send whose receiver was dropped + returns `Err`, which the recording fixture ignores (`let _ = self.tx.send(..)`) + — the receiver outlives the run in every test. + +## Testing strategy + +A test-local `Recorder` fixture (in `harness.rs`'s `#[cfg(test)] mod tests`, +alongside the existing `Ohlcv` / `AsOfSum` / `BarrierSum` fixtures — C9: no +speculative library surface). It holds its declared input kinds and an +`mpsc::Sender<(Timestamp, Vec)>`. **Read-back is via the channel, not via +`Rc`/`RefCell`** — this keeps `aura-engine/src` free of the interior-mutability +constructs the purity invariant (C7) forbids, regardless of how the audit's +purity grep is scoped, and is a faithful miniature of a real out-of-graph +recording destination. Its `eval` reads each input slot's newest value into a +`Vec`, sends `(ctx.now(), row)`, returns `None`. + +Migration (existing tests, behaviour preserved): every firing/DAG test currently +asserting on `run`'s returned `Vec>>` via `observe` is +rewritten to tap the previously-observed node with a `Recorder` and assert on the +drained channel. Because recording is sparse (fired cycles only), the expected +vectors drop their `None` hold-rows and carry timestamps: +`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`, +`within_source_diamond_rejoin_barrier_fires`, +`mixed_a_and_b_or_combine_on_one_node`, `ohlcv_bundles_five_field_record`, +`edge_binds_single_field_high_minus_low`, `distinct_edges_read_distinct_fields`. +The `bootstrap_rejects_*` tests drop the `observe` argument; `bootstrap_rejects_a_cycle`, +`bootstrap_rejects_a_kind_mismatch`, `bootstrap_rejects_from_field_out_of_range`, +and `bootstrap_rejects_per_field_kind_mismatch` are otherwise unchanged. +`bootstrap_rejects_a_bad_index` currently triggers `BadIndex` via `observe 5` +(an out-of-range observe index) — with `observe` removed, its trigger no longer +exists, so it is **repurposed** to raise `BadIndex` via an out-of-range *edge* +target (e.g. `Edge { to: 9, .. }` into a two-node graph): `BadIndex` itself is +unchanged, only the path that reaches it. + +New proof tests (the #2 deliverable; the exhaustive M×N×K matrix stays #3): + +- `multi_sink_records_distinct_interior_streams` — two recorders tap SMA(2) and + SMA(4) in one run; both drained streams are individually correct. The headline + "one run records many streams". +- `recorder_taps_all_fields_of_a_record` — one recorder with five inputs taps all + five OHLCV fields via five edges; its recorded row is the whole bar (ties 0005 + in: recording a multi-field producer is N field-wise edges, no whole-record + bind). +- `recorder_records_mixed_scalar_kinds` — a recorder with i64 + f64 + bool + + timestamp inputs records a four-field mixed-kind row, proving recording is not + f64-only. +- `node_is_producer_and_sink_at_once` — a node returns an output forwarded to a + downstream consumer **and** records to a channel in the same `eval`; both the + forwarded computation and the recorded stream are correct. Proves the C8 "both" + role. +- `recording_is_deterministic` — two fresh harnesses, two channels, identical + input; the two drained streams are bit-identical (C1). +- Recorder firing modes: a `Firing::Barrier` recorder records only on + barrier-complete cycles; a `Firing::Any` recorder records on any-fresh — + asserted by the timestamps present in the drained stream. +- `bootstrap_rejects_kind_mismatched_recorder_edge` — the must-fail above. + +aura-core: a `ctx_now_returns_cycle_timestamp` unit test. + +Gates: `cargo build/test --workspace`, `cargo clippy --workspace --all-targets -D +warnings`, and the audit's surface-purity grep (no `dyn Any` / `Rc` / `RefCell` +in engine source — the channel read-back keeps it clean). + +## Acceptance criteria + +The project's feature-acceptance criterion (CLAUDE.md): the audience naturally +reaches for it; it measurably improves correctness/removes redundancy; it +reintroduces no failure class a core constraint exists to prevent. + +- **Naturally reached for.** The worked example above is the program a downstream + author writes to record more than one stream from a run — impossible before + (one `observe` index). #3's matrix consumes exactly this surface. +- **Improves correctness / removes redundancy.** One mechanism (a node recording + via a held destination) replaces the special-case `observe` index; the engine + surface shrinks (a field, a parameter, a return type, and a per-cycle branch + all removed). Multi-stream recording becomes structurally reachable. +- **Reintroduces no forbidden failure class.** Determinism (C1) holds — a + recorded side effect cannot feed back into the graph (no out-edge for recorded + values), and the determinism test re-runs bit-identical. Purity (C7) holds — + no `dyn Any`/`Rc`/`RefCell` in engine source; the engine gains no `dyn FnMut`. + Causality (C2) holds — `now()` is the present. Field-wise binding (C8/0005) is + untouched — recording a K-field producer is K edges. +- Ship gate: `observe` gone; `bootstrap` and `run` signatures as specified; + `Ctx::now()` present; the full test matrix green; the determinism re-run + bit-identical; the C8/C22 realization notes added to the ledger; all three + gates (build/test, clippy, purity grep) clean.