feat(aura-std): tap subscriber sinks — TapFold, TapRecorder, TapLive

The three C8 sinks realizing the per-tap subscription model (#283): TapFold
(closed FoldKind vocabulary Count|Sum|Mean|Min|Max|First|Last, in-graph
accumulation, exactly one finalize row), TapRecorder (bounded
SyncSender<(Timestamp, Cell)>, lossless backpressure — a full channel blocks
the sim, wall time never state), and TapLive (consumer-owned closure, inline
on the sim thread). All three move (Timestamp, Cell) — 16 B, Copy — instead
of (Timestamp, Vec<Scalar>): zero per-cycle heap on the tap path (#77 part
2). The shared kind-dispatched newest-cell read lives in tap_cell.rs.

Known-good subset commit of iter tap-subscribers (tasks 1-3 of 7, each gated
implementer -> spec-compliance -> quality DONE; the loop blocked on a task-4
plan defect, repaired separately): the trace-store streaming writer, the
TapPlan wiring seam, and both entry-point rewires follow in this same cycle.
Verified directly: cargo test -p aura-std — 109 passed, 0 failed.

refs #283, refs #77

crates/aura-std/src/lib.rs

@@ -43,6 +43,10 @@ mod sign;
 mod sma;
 mod sqrt;
 mod sub;
+mod tap_cell;
+mod tap_fold;
+mod tap_live;
+mod tap_recorder;
 mod when;
 pub use abs::Abs;
 pub use add::Add;
@@ -70,4 +74,7 @@ pub use sign::Sign;
 pub use sma::Sma;
 pub use sqrt::Sqrt;
 pub use sub::Sub;
+pub use tap_fold::{fold_binds_at, fold_output_kind, FoldKind, TapFold};
+pub use tap_live::TapLive;
+pub use tap_recorder::TapRecorder;
 pub use when::When;

crates/aura-std/src/tap_cell.rs

@@ -0,0 +1,17 @@
+//! Shared single-column read for the tap sinks (`TapRecorder`, `TapFold`,
+//! `TapLive`): the kind-dispatched newest-value read `Recorder::eval` does,
+//! re-encoded as a bare `Cell` (the C7 kind-erased carrier) instead of a
+//! kind-tagged `Scalar` — the #77-part-2 zero-heap payload.
+
+use aura_core::{Cell, Ctx, ScalarKind};
+
+/// The newest value of input column 0, as a `Cell`. `None` until the column
+/// is warm (the `?` on the window read), mirroring `Recorder`'s warm-up.
+pub(crate) fn newest_cell(kind: ScalarKind, ctx: &Ctx<'_>) -> Option<Cell> {
+    Some(match kind {
+        ScalarKind::F64 => Cell::from_f64(ctx.f64_in(0).get(0)?),
+        ScalarKind::I64 => Cell::from_i64(ctx.i64_in(0).get(0)?),
+        ScalarKind::Bool => Cell::from_bool(ctx.bool_in(0).get(0)?),
+        ScalarKind::Timestamp => Cell::from_ts(ctx.ts_in(0).get(0)?),
+    })
+}

crates/aura-std/src/tap_fold.rs

@@ -0,0 +1,263 @@
+//! `TapFold` — a folding sink (C8) over one declared-tap column: accumulates
+//! owned per-cycle state (no channel traffic, no heap per cycle) and emits
+//! exactly one summary row on `finalize` (the #138 lifecycle; `SeriesReducer`
+//! precedent). The fold is named from the closed [`FoldKind`] vocabulary
+//! (C25: closed, typed; a new aggregate is a new named entry, never inline
+//! logic in a data artifact). C7-pure: owned state + the channel handle.
+
+use aura_core::{Cell, Ctx, Node, Scalar, ScalarKind, Timestamp};
+use std::sync::mpsc::Sender;
+
+use crate::tap_cell::newest_cell;
+
+/// The closed fold vocabulary. Bind rules: `Sum`/`Mean`/`Min`/`Max` bind only
+/// at f64 taps ([`fold_binds_at`]); `Count` at any kind; `First`/`Last`
+/// kind-preserving at any kind. Output kinds: [`fold_output_kind`].
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum FoldKind {
+    Count,
+    Sum,
+    Mean,
+    Min,
+    Max,
+    First,
+    Last,
+}
+
+/// May `fold` bind at a tap of column kind `kind`? The arithmetic folds are
+/// f64-only; `Count`/`First`/`Last` are kind-agnostic.
+pub fn fold_binds_at(fold: FoldKind, kind: ScalarKind) -> bool {
+    match fold {
+        FoldKind::Sum | FoldKind::Mean | FoldKind::Min | FoldKind::Max => {
+            kind == ScalarKind::F64
+        }
+        FoldKind::Count | FoldKind::First | FoldKind::Last => true,
+    }
+}
+
+/// The output column kind of `fold` at a tap of column kind `kind`:
+/// `Count` → i64; the arithmetic folds → f64; `First`/`Last` preserve `kind`.
+pub fn fold_output_kind(fold: FoldKind, kind: ScalarKind) -> ScalarKind {
+    match fold {
+        FoldKind::Count => ScalarKind::I64,
+        FoldKind::Sum | FoldKind::Mean | FoldKind::Min | FoldKind::Max => ScalarKind::F64,
+        FoldKind::First | FoldKind::Last => kind,
+    }
+}
+
+/// Owned accumulator for every fold kind at once (a handful of words; keeping
+/// it total avoids a per-kind state enum). Sequential `sum += v` matches a
+/// slice-driven mean's operation order, so streamed and slice `Mean` agree
+/// bit-for-bit (the `SeriesFold` discipline, IEEE-754 non-associativity
+/// respected).
+#[derive(Clone, Copy, Debug)]
+struct FoldState {
+    count: u64,
+    sum: f64,
+    min: f64,
+    max: f64,
+    first: Option<(Timestamp, Cell)>,
+    last: Option<Cell>,
+}
+
+impl FoldState {
+    fn new() -> Self {
+        Self {
+            count: 0,
+            sum: 0.0,
+            min: f64::INFINITY,
+            max: f64::NEG_INFINITY,
+            first: None,
+            last: None,
+        }
+    }
+
+    fn fold(&mut self, ts: Timestamp, cell: Cell, kind: ScalarKind) {
+        self.count += 1;
+        if kind == ScalarKind::F64 {
+            let v = cell.f64();
+            self.sum += v;
+            if v < self.min {
+                self.min = v;
+            }
+            if v > self.max {
+                self.max = v;
+            }
+        }
+        if self.first.is_none() {
+            self.first = Some((ts, cell));
+        }
+        self.last = Some(cell);
+    }
+
+    fn folded_any(&self) -> bool {
+        self.count > 0
+    }
+
+    /// The summary value; call only when `folded_any()`.
+    fn output(&self, fold: FoldKind, kind: ScalarKind) -> Scalar {
+        match fold {
+            FoldKind::Count => Scalar::i64(self.count as i64),
+            FoldKind::Sum => Scalar::f64(self.sum),
+            FoldKind::Mean => Scalar::f64(self.sum / self.count as f64),
+            FoldKind::Min => Scalar::f64(self.min),
+            FoldKind::Max => Scalar::f64(self.max),
+            FoldKind::First => {
+                Scalar::from_cell(kind, self.first.expect("folded_any checked").1)
+            }
+            FoldKind::Last => Scalar::from_cell(kind, self.last.expect("folded_any checked")),
+        }
+    }
+}
+
+/// A single-column folding sink. Emits exactly one row on `finalize` (never
+/// twice, never on a never-warm tap): `(row_ts, [output])`, where `row_ts` is
+/// the last folded value's ts — except `First`, which carries the first warm
+/// value's ts (the instant its reported value was determined).
+pub struct TapFold {
+    kind: ScalarKind,
+    fold: FoldKind,
+    state: FoldState,
+    last_ts: Timestamp,
+    emitted: bool,
+    tx: Sender<(Timestamp, Vec<Scalar>)>,
+}
+
+impl TapFold {
+    pub fn new(kind: ScalarKind, fold: FoldKind, tx: Sender<(Timestamp, Vec<Scalar>)>) -> Self {
+        debug_assert!(fold_binds_at(fold, kind), "caller validates fold/kind before binding");
+        Self { kind, fold, state: FoldState::new(), last_ts: Timestamp(0), emitted: false, tx }
+    }
+}
+
+impl Node for TapFold {
+    fn lookbacks(&self) -> Vec<usize> {
+        vec![1]
+    }
+
+    fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> {
+        let cell = newest_cell(self.kind, &ctx)?; // None until warm
+        self.state.fold(ctx.now(), cell, self.kind);
+        self.last_ts = ctx.now();
+        None
+    }
+
+    fn finalize(&mut self) {
+        if self.state.folded_any() && !self.emitted {
+            self.emitted = true;
+            let ts = match self.fold {
+                FoldKind::First => self.state.first.expect("folded_any checked").0,
+                _ => self.last_ts,
+            };
+            let _ = self.tx.send((ts, vec![self.state.output(self.fold, self.kind)]));
+        }
+    }
+
+    fn label(&self) -> String {
+        format!("TapFold({:?})", self.fold)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use aura_core::AnyColumn;
+    use std::sync::mpsc;
+
+    /// Drive a fold over an f64 series and return the emitted rows.
+    fn run_f64_fold(fold: FoldKind, values: &[f64]) -> Vec<(Timestamp, Vec<Scalar>)> {
+        let (tx, rx) = mpsc::channel();
+        let mut sink = TapFold::new(ScalarKind::F64, fold, tx);
+        let mut inputs = vec![AnyColumn::with_capacity(ScalarKind::F64, 1)];
+        for (i, &v) in values.iter().enumerate() {
+            inputs[0].push(Scalar::f64(v)).unwrap();
+            assert_eq!(sink.eval(Ctx::new(&inputs, Timestamp(i as i64 + 1))), None);
+        }
+        sink.finalize();
+        rx.try_iter().collect()
+    }
+
+    #[test]
+    fn mean_matches_slice_mean_bit_for_bit() {
+        let values = [0.1, 0.2, 0.30000000000000004, -1.5, 2.75];
+        let rows = run_f64_fold(FoldKind::Mean, &values);
+        let slice_mean = values.iter().sum::<f64>() / values.len() as f64;
+        assert_eq!(rows, vec![(Timestamp(5), vec![Scalar::f64(slice_mean)])]);
+    }
+
+    #[test]
+    fn count_sum_min_max_over_a_known_series() {
+        let values = [3.0, -1.0, 2.0];
+        assert_eq!(
+            run_f64_fold(FoldKind::Count, &values),
+            vec![(Timestamp(3), vec![Scalar::i64(3)])]
+        );
+        assert_eq!(
+            run_f64_fold(FoldKind::Sum, &values),
+            vec![(Timestamp(3), vec![Scalar::f64(4.0)])]
+        );
+        assert_eq!(
+            run_f64_fold(FoldKind::Min, &values),
+            vec![(Timestamp(3), vec![Scalar::f64(-1.0)])]
+        );
+        assert_eq!(
+            run_f64_fold(FoldKind::Max, &values),
+            vec![(Timestamp(3), vec![Scalar::f64(3.0)])]
+        );
+    }
+
+    #[test]
+    fn first_carries_the_first_warm_ts_and_last_the_last() {
+        let values = [7.0, 8.0, 9.0];
+        assert_eq!(
+            run_f64_fold(FoldKind::First, &values),
+            vec![(Timestamp(1), vec![Scalar::f64(7.0)])],
+            "First reports the first warm value at ITS ts"
+        );
+        assert_eq!(
+            run_f64_fold(FoldKind::Last, &values),
+            vec![(Timestamp(3), vec![Scalar::f64(9.0)])]
+        );
+    }
+
+    #[test]
+    fn never_warm_tap_emits_nothing_and_finalize_emits_once() {
+        let (tx, rx) = mpsc::channel();
+        let mut sink = TapFold::new(ScalarKind::F64, FoldKind::Count, tx);
+        let inputs = vec![AnyColumn::with_capacity(ScalarKind::F64, 1)];
+        // cold eval, then finalize: nothing emitted.
+        assert_eq!(sink.eval(Ctx::new(&inputs, Timestamp(1))), None);
+        sink.finalize();
+        assert!(rx.try_recv().is_err(), "a never-warm fold emits no row");
+
+        // warm one value; a double finalize still emits exactly one row.
+        let (tx2, rx2) = mpsc::channel();
+        let mut sink2 = TapFold::new(ScalarKind::I64, FoldKind::Last, tx2);
+        let mut inputs2 = vec![AnyColumn::with_capacity(ScalarKind::I64, 1)];
+        inputs2[0].push(Scalar::i64(42)).unwrap();
+        assert_eq!(sink2.eval(Ctx::new(&inputs2, Timestamp(9))), None);
+        sink2.finalize();
+        sink2.finalize();
+        let rows: Vec<_> = rx2.try_iter().collect();
+        assert_eq!(rows, vec![(Timestamp(9), vec![Scalar::i64(42)])]);
+    }
+
+    #[test]
+    fn bind_rules_and_output_kinds_are_the_specced_table() {
+        for fold in [FoldKind::Sum, FoldKind::Mean, FoldKind::Min, FoldKind::Max] {
+            assert!(fold_binds_at(fold, ScalarKind::F64));
+            assert!(!fold_binds_at(fold, ScalarKind::I64));
+            assert!(!fold_binds_at(fold, ScalarKind::Bool));
+            assert!(!fold_binds_at(fold, ScalarKind::Timestamp));
+            assert_eq!(fold_output_kind(fold, ScalarKind::F64), ScalarKind::F64);
+        }
+        for kind in [ScalarKind::F64, ScalarKind::I64, ScalarKind::Bool, ScalarKind::Timestamp] {
+            assert!(fold_binds_at(FoldKind::Count, kind));
+            assert!(fold_binds_at(FoldKind::First, kind));
+            assert!(fold_binds_at(FoldKind::Last, kind));
+            assert_eq!(fold_output_kind(FoldKind::Count, kind), ScalarKind::I64);
+            assert_eq!(fold_output_kind(FoldKind::First, kind), kind);
+            assert_eq!(fold_output_kind(FoldKind::Last, kind), kind);
+        }
+    }
+}

crates/aura-std/src/tap_live.rs

@@ -0,0 +1,88 @@
+//! `TapLive` — the live-subscription sink (C8) for one declared tap: calls a
+//! consumer-owned closure `(Timestamp, Cell)` inline per fired warm cycle, on
+//! the sim thread, zero alloc. Loss policy is the closure's own (a UI
+//! consumer typically `try_send`s into its own channel and drops when behind)
+//! — per #283: consumer-owned logic and state, no registry ceremony.
+
+use aura_core::{Cell, Ctx, Node, ScalarKind, Timestamp};
+
+use crate::tap_cell::newest_cell;
+
+/// A single-column live sink around a consumer closure.
+pub struct TapLive {
+    kind: ScalarKind,
+    consumer: Box<dyn FnMut(Timestamp, Cell) + Send>,
+}
+
+impl TapLive {
+    pub fn new(kind: ScalarKind, consumer: impl FnMut(Timestamp, Cell) + Send + 'static) -> Self {
+        Self { kind, consumer: Box::new(consumer) }
+    }
+}
+
+impl Node for TapLive {
+    fn lookbacks(&self) -> Vec<usize> {
+        vec![1]
+    }
+
+    fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> {
+        let cell = newest_cell(self.kind, &ctx)?; // None until warm
+        (self.consumer)(ctx.now(), cell);
+        None
+    }
+
+    fn label(&self) -> String {
+        "TapLive".to_string()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use aura_core::{AnyColumn, Scalar};
+    use std::sync::mpsc;
+
+    #[test]
+    fn calls_the_consumer_per_warm_cycle_in_order_and_not_before() {
+        let (tx, rx) = mpsc::channel();
+        let mut sink = TapLive::new(ScalarKind::F64, move |ts, cell| {
+            let _ = tx.send((ts, cell.f64()));
+        });
+        let mut inputs = vec![AnyColumn::with_capacity(ScalarKind::F64, 1)];
+
+        // cold: consumer not called.
+        assert_eq!(sink.eval(Ctx::new(&inputs, Timestamp(1))), None);
+        assert!(rx.try_recv().is_err(), "no call before warm-up");
+
+        for (t, v) in [(2_i64, 1.5_f64), (3, 2.5)] {
+            inputs[0].push(Scalar::f64(v)).unwrap();
+            assert_eq!(sink.eval(Ctx::new(&inputs, Timestamp(t))), None);
+        }
+        let got: Vec<(Timestamp, f64)> = rx.try_iter().collect();
+        assert_eq!(got, vec![(Timestamp(2), 1.5), (Timestamp(3), 2.5)]);
+    }
+
+    /// `newest_cell`'s four-kind dispatch (shared with `TapFold`) is
+    /// otherwise only exercised at F64/I64; the Bool and Timestamp arms
+    /// need a witness too, so a regression in either goes caught.
+    #[test]
+    fn calls_the_consumer_at_bool_and_timestamp_kinds_too() {
+        let (tx, rx) = mpsc::channel();
+        let mut sink = TapLive::new(ScalarKind::Bool, move |ts, cell| {
+            let _ = tx.send((ts, cell.bool()));
+        });
+        let mut inputs = vec![AnyColumn::with_capacity(ScalarKind::Bool, 1)];
+        inputs[0].push(Scalar::bool(true)).unwrap();
+        assert_eq!(sink.eval(Ctx::new(&inputs, Timestamp(1))), None);
+        assert_eq!(rx.try_recv(), Ok((Timestamp(1), true)));
+
+        let (tx2, rx2) = mpsc::channel();
+        let mut sink2 = TapLive::new(ScalarKind::Timestamp, move |ts, cell| {
+            let _ = tx2.send((ts, cell.ts()));
+        });
+        let mut inputs2 = vec![AnyColumn::with_capacity(ScalarKind::Timestamp, 1)];
+        inputs2[0].push(Scalar::ts(Timestamp(99))).unwrap();
+        assert_eq!(sink2.eval(Ctx::new(&inputs2, Timestamp(1))), None);
+        assert_eq!(rx2.try_recv(), Ok((Timestamp(1), Timestamp(99))));
+    }
+}

crates/aura-std/src/tap_recorder.rs

@@ -0,0 +1,96 @@
+//! `TapRecorder` — the record-subscription sink (C8) for one declared tap:
+//! sends `(ctx.now(), cell)` — 16 B, `Copy`, zero heap per cycle (#77 part 2)
+//! — on a caller-chosen **bounded** `SyncSender`. Lossless by contract: a
+//! full channel blocks the sim (wall time only, never state — the consumer
+//! is one-way, C1 intact); a disconnected receiver returns `Err` immediately
+//! and the run completes (the failure surfaces at the caller's join/finish).
+//! C7-pure: no interior mutability, only the owned channel handle.
+
+use aura_core::{Cell, Ctx, Node, ScalarKind, Timestamp};
+use std::sync::mpsc::SyncSender;
+
+use crate::tap_cell::newest_cell;
+
+/// A single-column recording sink over a bounded channel. `None` until the
+/// column is warm (records nothing), `None` always (pure consumer).
+pub struct TapRecorder {
+    kind: ScalarKind,
+    tx: SyncSender<(Timestamp, Cell)>,
+}
+
+impl TapRecorder {
+    pub fn new(kind: ScalarKind, tx: SyncSender<(Timestamp, Cell)>) -> Self {
+        Self { kind, tx }
+    }
+}
+
+impl Node for TapRecorder {
+    fn lookbacks(&self) -> Vec<usize> {
+        vec![1]
+    }
+
+    fn eval(&mut self, ctx: Ctx<'_>) -> Option<&[Cell]> {
+        let cell = newest_cell(self.kind, &ctx)?; // None until warm
+        // Blocks when the writer is behind (lossless); Err on a dead
+        // receiver — ignored here, surfaced by the caller at join.
+        let _ = self.tx.send((ctx.now(), cell));
+        None
+    }
+
+    fn label(&self) -> String {
+        "TapRecorder".to_string()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use aura_core::{AnyColumn, Scalar};
+    use std::sync::mpsc;
+
+    #[test]
+    fn records_ts_cell_pairs_after_warmup() {
+        let (tx, rx) = mpsc::sync_channel(8);
+        let mut sink = TapRecorder::new(ScalarKind::F64, tx);
+        let mut inputs = vec![AnyColumn::with_capacity(ScalarKind::F64, 1)];
+
+        // cold: nothing recorded.
+        assert_eq!(sink.eval(Ctx::new(&inputs, Timestamp(1))), None);
+        assert!(rx.try_recv().is_err());
+
+        for (t, v) in [(2_i64, 10.0_f64), (3, 20.0), (4, 30.0)] {
+            inputs[0].push(Scalar::f64(v)).unwrap();
+            assert_eq!(sink.eval(Ctx::new(&inputs, Timestamp(t))), None);
+        }
+        let got: Vec<(Timestamp, f64)> = rx.try_iter().map(|(t, c)| (t, c.f64())).collect();
+        assert_eq!(
+            got,
+            vec![(Timestamp(2), 10.0), (Timestamp(3), 20.0), (Timestamp(4), 30.0)]
+        );
+    }
+
+    /// Backpressure losslessness: a capacity-1 channel with a concurrent
+    /// consumer never drops a value — the producer blocks instead. 100 sends
+    /// through a 1-slot channel arrive complete and in order.
+    #[test]
+    fn capacity_one_channel_is_lossless_under_backpressure() {
+        let (tx, rx) = mpsc::sync_channel::<(Timestamp, Cell)>(1);
+        let consumer = std::thread::spawn(move || {
+            let mut got = Vec::new();
+            for (_, cell) in rx.iter() {
+                got.push(cell.i64());
+            }
+            got
+        });
+        {
+            let mut sink = TapRecorder::new(ScalarKind::I64, tx);
+            let mut inputs = vec![AnyColumn::with_capacity(ScalarKind::I64, 1)];
+            for i in 0..100_i64 {
+                inputs[0].push(Scalar::i64(i)).unwrap();
+                assert_eq!(sink.eval(Ctx::new(&inputs, Timestamp(i + 1))), None);
+            }
+        } // sink (and tx) dropped → consumer's iter ends
+        let got = consumer.join().expect("consumer thread");
+        assert_eq!(got, (0..100).collect::<Vec<i64>>(), "lossless, in order");
+    }
+}