spec: 0006 sink recording

A sink is a role, not a type: recording is an out-of-graph side effect a node performs in eval, pushing to a destination it holds as a field. No Sink type, no trait, no engine sink-awareness. The engine stays the router for in-graph routing (edge table = graph-as-data); the node-side push is the only out-of-graph escape, and that boundary is the determinism boundary. A node may be producer and sink at once (C8 both). Engine surface shrinks: Ctx gains now(); the observe index (field, bounds check, per-cycle collection) is removed; bootstrap drops its 4th param; run returns (). Recording streams are sparse and timestamped (only fired cycles), replacing the dense Vec<Option<row>>. Realization notes land on C8 and C22; no new contract. refs #2
2026-06-04 14:11:00 +02:00
parent 57310b83c7
commit 22ee99abcd
1 changed files with 330 additions and 0 deletions
@@ -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<Scalar>)> = rx_fast.try_iter().collect();
+let slow: Vec<(Timestamp, Vec<Scalar>)> = 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<Harness, BootstrapError>
+pub fn run(&mut self, streams) -> Vec<Option<Vec<Scalar>>>
+
+// after
+pub struct Harness { nodes, topo, out_edges, sources }   // no observe
+pub fn bootstrap(nodes, sources, edges) -> Result<Harness, BootstrapError>  // 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.<kind>_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<Option<row>>`
+(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<Scalar>)>`. **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<Scalar>`, sends `(ctx.now(), row)`, returns `None`.
+
+Migration (existing tests, behaviour preserved): every firing/DAG test currently
+asserting on `run`'s returned `Vec<Option<Vec<Scalar>>>` 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.