feat(core): the streaming substrate — scalars, columns, type-erased edges

Cycle 0001 of the walking-skeleton milestone: the aura-core data substrate
everything streams through.

- scalar.rs: the closed four-type set — Scalar (Copy POD carrier), ScalarKind
  (the kind tag), Timestamp (i64 epoch-ns newtype). No boxed Value, no dyn Any,
  no heap (C7).
- column.rs: Column<T>, a fixed-capacity bounded-lookback ring pre-sized at
  construction (no hot-loop realloc, C8), financial-indexed (index 0 = newest),
  carrying a monotonic run_count freshness primitive (C5). Window<'_,T> is its
  zero-copy read view with newest-at-0 Index. No RefCell/Rc — a sim is
  single-threaded (C1), so push is &mut and read is &.
- any.rs: AnyColumn, the type-erased edge over the four Column kinds. push()
  kind-checks at the edge (wiring time) and rejects a wrong-kind value with
  KindMismatch, leaving the column untouched; the hot path bypasses the check
  via the monomorphic as_*_mut accessors (C7).
- error.rs: KindMismatch { expected, got }, the guard error.

Sharpens the RustAst rtl/ reference (VecDeque + RefCell + boxed Value) into a
fixed ring of Copy scalars with direct dispatch. Node trait, Ctx, firing
policies, and the sim loop are deferred to subsequent cycles.

Verified independently: cargo build --workspace (0 warnings), cargo test
--workspace (14 aura-core tests green incl. ring-wrap, run_count monotonicity,
zero-copy window, and the wrong_kind_push_is_rejected guard), cargo clippy
--workspace --all-targets -D warnings (clean), and a surface-purity grep
confirming no RefCell/Rc/dyn Any on the substrate.

refs #0001-core-streaming-substrate

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

crates/aura-core/src/any.rs

@@ -0,0 +1,170 @@
+//! `AnyColumn` — the type-erased edge: the four concrete `Column<T>` behind one
+//! enum. The kind check is paid at the edge (wiring time); the hot loop grabs
+//! the concrete column once (`as_*_mut`) and pushes monomorphically (C7).
+
+use crate::column::Column;
+use crate::error::KindMismatch;
+use crate::scalar::{Scalar, ScalarKind, Timestamp};
+
+/// A type-erased edge over the four scalar columns (C7).
+pub enum AnyColumn {
+    I64(Column<i64>),
+    F64(Column<f64>),
+    Bool(Column<bool>),
+    Ts(Column<Timestamp>),
+}
+
+impl AnyColumn {
+    /// Create a type-erased column of `kind`, sized to `lookback` (C8).
+    pub fn with_capacity(kind: ScalarKind, lookback: usize) -> Self {
+        match kind {
+            ScalarKind::I64 => AnyColumn::I64(Column::with_capacity(lookback)),
+            ScalarKind::F64 => AnyColumn::F64(Column::with_capacity(lookback)),
+            ScalarKind::Bool => AnyColumn::Bool(Column::with_capacity(lookback)),
+            ScalarKind::Timestamp => AnyColumn::Ts(Column::with_capacity(lookback)),
+        }
+    }
+
+    pub fn kind(&self) -> ScalarKind {
+        match self {
+            AnyColumn::I64(_) => ScalarKind::I64,
+            AnyColumn::F64(_) => ScalarKind::F64,
+            AnyColumn::Bool(_) => ScalarKind::Bool,
+            AnyColumn::Ts(_) => ScalarKind::Timestamp,
+        }
+    }
+
+    pub fn len(&self) -> usize {
+        match self {
+            AnyColumn::I64(c) => c.len(),
+            AnyColumn::F64(c) => c.len(),
+            AnyColumn::Bool(c) => c.len(),
+            AnyColumn::Ts(c) => c.len(),
+        }
+    }
+
+    pub fn is_empty(&self) -> bool {
+        self.len() == 0
+    }
+
+    pub fn run_count(&self) -> u64 {
+        match self {
+            AnyColumn::I64(c) => c.run_count(),
+            AnyColumn::F64(c) => c.run_count(),
+            AnyColumn::Bool(c) => c.run_count(),
+            AnyColumn::Ts(c) => c.run_count(),
+        }
+    }
+
+    /// Edge-time typed push. `Err(KindMismatch)` if `v`'s kind != this column's
+    /// kind; the column is left untouched (C7 guard). The hot path bypasses this
+    /// by taking the concrete `Column<T>` once via `as_*_mut`.
+    pub fn push(&mut self, v: Scalar) -> Result<(), KindMismatch> {
+        match (self, v) {
+            (AnyColumn::I64(c), Scalar::I64(x)) => {
+                c.push(x);
+                Ok(())
+            }
+            (AnyColumn::F64(c), Scalar::F64(x)) => {
+                c.push(x);
+                Ok(())
+            }
+            (AnyColumn::Bool(c), Scalar::Bool(x)) => {
+                c.push(x);
+                Ok(())
+            }
+            (AnyColumn::Ts(c), Scalar::Ts(x)) => {
+                c.push(x);
+                Ok(())
+            }
+            (this, v) => Err(KindMismatch { expected: this.kind(), got: v.kind() }),
+        }
+    }
+
+    /// Read one type-erased value (0 = newest). `None` if cold / out of range.
+    pub fn get(&self, k: usize) -> Option<Scalar> {
+        match self {
+            AnyColumn::I64(c) => c.get(k).map(Scalar::I64),
+            AnyColumn::F64(c) => c.get(k).map(Scalar::F64),
+            AnyColumn::Bool(c) => c.get(k).map(Scalar::Bool),
+            AnyColumn::Ts(c) => c.get(k).map(Scalar::Ts),
+        }
+    }
+
+    /// Concrete-column accessor for the monomorphic hot path; `Some` only for
+    /// the matching kind.
+    pub fn as_i64_mut(&mut self) -> Option<&mut Column<i64>> {
+        match self {
+            AnyColumn::I64(c) => Some(c),
+            _ => None,
+        }
+    }
+
+    pub fn as_f64_mut(&mut self) -> Option<&mut Column<f64>> {
+        match self {
+            AnyColumn::F64(c) => Some(c),
+            _ => None,
+        }
+    }
+
+    pub fn as_bool_mut(&mut self) -> Option<&mut Column<bool>> {
+        match self {
+            AnyColumn::Bool(c) => Some(c),
+            _ => None,
+        }
+    }
+
+    pub fn as_ts_mut(&mut self) -> Option<&mut Column<Timestamp>> {
+        match self {
+            AnyColumn::Ts(c) => Some(c),
+            _ => None,
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn same_kind_push_round_trips() {
+        let mut e = AnyColumn::with_capacity(ScalarKind::F64, 4);
+        e.push(Scalar::F64(1.5)).unwrap();
+        e.push(Scalar::F64(2.5)).unwrap();
+        assert_eq!(e.kind(), ScalarKind::F64);
+        assert_eq!(e.len(), 2);
+        assert_eq!(e.run_count(), 2);
+        assert_eq!(e.get(0), Some(Scalar::F64(2.5))); // newest
+        assert_eq!(e.get(1), Some(Scalar::F64(1.5)));
+        assert_eq!(e.get(2), None);
+    }
+
+    #[test]
+    fn wrong_kind_push_is_rejected() {
+        let mut edge = AnyColumn::with_capacity(ScalarKind::I64, 8);
+        let err = edge.push(Scalar::F64(1.5)).unwrap_err(); // f64 into an i64 edge
+        assert_eq!(err, KindMismatch { expected: ScalarKind::I64, got: ScalarKind::F64 });
+        assert_eq!(edge.len(), 0); // nothing was stored
+        assert_eq!(edge.run_count(), 0); // and the counter did not move
+    }
+
+    #[test]
+    fn concrete_accessor_matches_only_its_kind() {
+        let mut e = AnyColumn::with_capacity(ScalarKind::F64, 2);
+        assert!(e.as_f64_mut().is_some());
+        assert!(e.as_i64_mut().is_none());
+        assert!(e.as_bool_mut().is_none());
+        assert!(e.as_ts_mut().is_none());
+        // hot-path push through the concrete column, monomorphic:
+        e.as_f64_mut().unwrap().push(3.0);
+        assert_eq!(e.get(0), Some(Scalar::F64(3.0)));
+        assert_eq!(e.run_count(), 1);
+    }
+
+    #[test]
+    fn timestamp_edge_round_trips() {
+        let mut e = AnyColumn::with_capacity(ScalarKind::Timestamp, 2);
+        e.push(Scalar::Ts(Timestamp(1_700_000_000_000_000_000))).unwrap();
+        assert_eq!(e.get(0), Some(Scalar::Ts(Timestamp(1_700_000_000_000_000_000))));
+    }
+}

crates/aura-core/src/column.rs

@@ -0,0 +1,176 @@
+//! `Column<T>` — a fixed-capacity, bounded-lookback SoA ring with financial
+//! indexing (index 0 = newest), and its zero-copy read view `Window` (C5/C8).
+
+/// A fixed-capacity bounded-lookback column. Pre-sized at construction
+/// (C8: no realloc in the hot loop); index 0 = newest; carries a monotonic
+/// `run_count` of total pushes — the freshness primitive (C5).
+pub struct Column<T> {
+    buf: Box<[T]>,
+    head: usize,    // next write slot
+    len: usize,     // valid elements, saturates at capacity
+    run_count: u64, // monotonic total pushes (C5)
+}
+
+impl<T: Copy + Default> Column<T> {
+    /// Create a column sized to `lookback` (must be >= 1). The engine sizes this
+    /// once at wiring; it is never reallocated (C8).
+    pub fn with_capacity(lookback: usize) -> Self {
+        assert!(lookback >= 1, "Column lookback must be >= 1");
+        Column {
+            buf: vec![T::default(); lookback].into_boxed_slice(),
+            head: 0,
+            len: 0,
+            run_count: 0,
+        }
+    }
+
+    /// Append the newest value. O(1), allocation-free. On overflow the oldest
+    /// value is overwritten (true ring). Bumps `run_count`.
+    pub fn push(&mut self, v: T) {
+        let cap = self.buf.len();
+        self.buf[self.head] = v;
+        self.head = (self.head + 1) % cap;
+        if self.len < cap {
+            self.len += 1;
+        }
+        self.run_count += 1;
+    }
+
+    /// Financial index: 0 = newest, `len()-1` = oldest. `None` if `k >= len`
+    /// (C8: not-yet-warmed-up / out of range). Zero-copy.
+    pub fn get(&self, k: usize) -> Option<T> {
+        if k >= self.len {
+            return None;
+        }
+        let cap = self.buf.len();
+        Some(self.buf[(self.head + cap - 1 - k) % cap])
+    }
+
+    pub fn len(&self) -> usize {
+        self.len
+    }
+
+    pub fn is_empty(&self) -> bool {
+        self.len == 0
+    }
+
+    pub fn capacity(&self) -> usize {
+        self.buf.len()
+    }
+
+    /// The monotonic push counter — the freshness primitive (C5). Never moved
+    /// by a read.
+    pub fn run_count(&self) -> u64 {
+        self.run_count
+    }
+
+    /// A zero-copy read view handed to a reading node (C8).
+    pub fn window(&self) -> Window<'_, T> {
+        Window { col: self }
+    }
+}
+
+/// Read-only, zero-copy view into a [`Column`] — what a reading node sees (C8).
+/// Carries no storage, only a borrow.
+pub struct Window<'a, T> {
+    col: &'a Column<T>,
+}
+
+impl<'a, T: Copy + Default> Window<'a, T> {
+    /// Financial index, 0 = newest; `None` if cold / out of range.
+    pub fn get(&self, k: usize) -> Option<T> {
+        self.col.get(k)
+    }
+
+    pub fn len(&self) -> usize {
+        self.col.len()
+    }
+
+    pub fn is_empty(&self) -> bool {
+        self.col.is_empty()
+    }
+
+    pub fn run_count(&self) -> u64 {
+        self.col.run_count()
+    }
+}
+
+/// `window[k]` — newest-at-0 indexed access; panics on out-of-range like a
+/// slice. Nodes that may be cold use [`Window::get`] instead.
+impl<'a, T: Copy + Default> core::ops::Index<usize> for Window<'a, T> {
+    type Output = T;
+
+    fn index(&self, k: usize) -> &T {
+        let col = self.col;
+        assert!(k < col.len, "Window index {k} out of range (len {})", col.len);
+        let cap = col.buf.len();
+        &col.buf[(col.head + cap - 1 - k) % cap]
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn newest_at_zero_oldest_at_len_minus_one() {
+        let mut c = Column::<i64>::with_capacity(4);
+        c.push(10);
+        c.push(20);
+        c.push(30);
+        assert_eq!(c.len(), 3);
+        assert_eq!(c.get(0), Some(30)); // newest
+        assert_eq!(c.get(1), Some(20));
+        assert_eq!(c.get(2), Some(10)); // oldest
+        assert_eq!(c.get(3), None); // beyond len
+    }
+
+    #[test]
+    fn wraparound_keeps_newest_capacity_values() {
+        let mut c = Column::<i64>::with_capacity(3);
+        for v in [1, 2, 3, 4, 5] {
+            c.push(v);
+        }
+        assert_eq!(c.len(), 3); // saturated
+        assert_eq!(c.capacity(), 3);
+        assert_eq!(c.get(0), Some(5)); // newest
+        assert_eq!(c.get(1), Some(4));
+        assert_eq!(c.get(2), Some(3)); // oldest retained
+        assert_eq!(c.get(3), None); // 1 and 2 were dropped
+    }
+
+    #[test]
+    fn run_count_counts_pushes_not_len() {
+        let mut c = Column::<i64>::with_capacity(2);
+        assert_eq!(c.run_count(), 0);
+        for v in [1, 2, 3, 4, 5] {
+            c.push(v);
+        }
+        assert_eq!(c.run_count(), 5); // counts pushes, past the ring wrap
+        assert_eq!(c.len(), 2); // while len stays at capacity
+        let _ = c.get(0);
+        assert_eq!(c.run_count(), 5); // a read never moves it
+    }
+
+    #[test]
+    fn window_is_a_view_matching_the_column() {
+        let mut c = Column::<f64>::with_capacity(4);
+        c.push(1.0);
+        c.push(2.0);
+        let cap_before = c.capacity();
+        let w = c.window();
+        assert_eq!(w.len(), 2);
+        assert_eq!(w.get(0), Some(2.0));
+        assert_eq!(w[1], 1.0); // Index, newest-at-0
+        assert_eq!(w.run_count(), 2);
+        assert_eq!(c.capacity(), cap_before); // taking a window did not realloc
+    }
+
+    #[test]
+    fn empty_column_reads_none() {
+        let c = Column::<bool>::with_capacity(2);
+        assert!(c.is_empty());
+        assert_eq!(c.get(0), None);
+        assert_eq!(c.run_count(), 0);
+    }
+}

crates/aura-core/src/error.rs

@@ -0,0 +1,25 @@
+//! `KindMismatch` — the wiring-time guard error for a wrong-kind edge push (C7).
+
+use crate::scalar::ScalarKind;
+
+/// Returned by [`crate::AnyColumn::push`] when a `Scalar`'s kind does not match
+/// the column's kind; the column is left untouched.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub struct KindMismatch {
+    /// The kind the column holds.
+    pub expected: ScalarKind,
+    /// The kind of the value that was offered.
+    pub got: ScalarKind,
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn kind_mismatch_is_a_plain_value() {
+        let e = KindMismatch { expected: ScalarKind::I64, got: ScalarKind::F64 };
+        assert_eq!(e, KindMismatch { expected: ScalarKind::I64, got: ScalarKind::F64 });
+        assert_ne!(e, KindMismatch { expected: ScalarKind::F64, got: ScalarKind::I64 });
+    }
+}

crates/aura-core/src/lib.rs

@@ -1,18 +1,29 @@
 //! `aura-core` — the shared contract every aura component links against.
 //!
 //! This crate is the boundary forced by the cdylib hot-reload model: both the
-//! engine and every hot-reloadable node cdylib depend on it, so it must stay
-//! small and stable. It is the home of (to be specified):
+//! engine and every hot-reloadable node cdylib depend on it, so it stays small
+//! and stable.
 //!
-//! - the scalar base types streamed as SoA: `i64`, `f64`, `bool`, `timestamp`;
-//! - the `Node` trait — the contract an authored node implements (`schema` —
-//!   inputs, lookback, firing group, and tunable params with ranges — plus
-//!   `eval`), with engine-provided read-only input windows and node-internal
-//!   series;
-//! - the evaluation context `Ctx` (zero-copy indexed access into input columns,
-//!   fresh-mask, current-cycle timestamp);
-//! - the firing-policy declarations (A: fire-on-any-fresh + hold; B: all-fresh
-//!   barrier), declarable per input group.
+//! Delivered so far (cycle 0001 — the streaming substrate):
 //!
-//! Types are intentionally absent until the first spec lands under
-//! `docs/specs/`. This crate documents intent; it does not yet define API.
+//! - the four scalar base types streamed as SoA: [`Scalar`] / [`ScalarKind`] /
+//!   [`Timestamp`] (C7);
+//! - [`Column`] — a fixed-capacity, bounded-lookback ring with financial
+//!   indexing (index 0 = newest) and a `run_count` freshness primitive (C5/C8),
+//!   read zero-copy through a [`Window`];
+//! - [`AnyColumn`] — the type-erased edge over the four column kinds, with an
+//!   edge-time kind check ([`KindMismatch`]) (C7).
+//!
+//! Still to come (subsequent cycles): the `Node` trait and its `schema`/`eval`,
+//! the evaluation context `Ctx`, the firing policies (A: fire-on-any-fresh +
+//! hold; B: all-fresh barrier), and the deterministic sim loop.
+
+mod any;
+mod column;
+mod error;
+mod scalar;
+
+pub use any::AnyColumn;
+pub use column::{Column, Window};
+pub use error::KindMismatch;
+pub use scalar::{Scalar, ScalarKind, Timestamp};

crates/aura-core/src/scalar.rs

@@ -0,0 +1,95 @@
+//! The closed four-type scalar set streamed on aura's hot path (C7):
+//! `i64`, `f64`, `bool`, and `Timestamp` (epoch-ns UTC). The carrier `Scalar`
+//! is a `Copy` POD enum — no type-erased payloads, no heap, no reference counting.
+
+/// Canonical engine time: epoch-nanoseconds UTC. Newtype over `i64` (C7).
+/// `Default` (epoch 0) lets `Column<Timestamp>` pre-size its ring.
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, PartialOrd, Ord, Hash)]
+pub struct Timestamp(pub i64);
+
+/// The kind tag of a scalar / column, used for the edge-time type check (C7).
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
+pub enum ScalarKind {
+    I64,
+    F64,
+    Bool,
+    Timestamp,
+}
+
+/// The four scalar base types, type-erased into one `Copy` carrier (C7).
+#[derive(Clone, Copy, Debug, PartialEq)]
+pub enum Scalar {
+    I64(i64),
+    F64(f64),
+    Bool(bool),
+    Ts(Timestamp),
+}
+
+impl Scalar {
+    /// The kind tag of this scalar.
+    pub fn kind(self) -> ScalarKind {
+        match self {
+            Scalar::I64(_) => ScalarKind::I64,
+            Scalar::F64(_) => ScalarKind::F64,
+            Scalar::Bool(_) => ScalarKind::Bool,
+            Scalar::Ts(_) => ScalarKind::Timestamp,
+        }
+    }
+}
+
+impl From<i64> for Scalar {
+    fn from(v: i64) -> Self {
+        Scalar::I64(v)
+    }
+}
+impl From<f64> for Scalar {
+    fn from(v: f64) -> Self {
+        Scalar::F64(v)
+    }
+}
+impl From<bool> for Scalar {
+    fn from(v: bool) -> Self {
+        Scalar::Bool(v)
+    }
+}
+impl From<Timestamp> for Scalar {
+    fn from(v: Timestamp) -> Self {
+        Scalar::Ts(v)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn kind_matches_variant() {
+        assert_eq!(Scalar::I64(1).kind(), ScalarKind::I64);
+        assert_eq!(Scalar::F64(1.0).kind(), ScalarKind::F64);
+        assert_eq!(Scalar::Bool(true).kind(), ScalarKind::Bool);
+        assert_eq!(Scalar::Ts(Timestamp(1)).kind(), ScalarKind::Timestamp);
+    }
+
+    #[test]
+    fn from_widenings() {
+        assert_eq!(Scalar::from(7i64), Scalar::I64(7));
+        assert_eq!(Scalar::from(2.5f64), Scalar::F64(2.5));
+        assert_eq!(Scalar::from(true), Scalar::Bool(true));
+        assert_eq!(Scalar::from(Timestamp(9)), Scalar::Ts(Timestamp(9)));
+    }
+
+    #[test]
+    fn timestamp_orders_and_defaults() {
+        assert!(Timestamp(1) < Timestamp(2));
+        assert_eq!(Timestamp::default(), Timestamp(0));
+    }
+
+    #[test]
+    fn scalar_is_copy() {
+        let s = Scalar::F64(1.0);
+        let a = s;
+        let b = s; // Copy: `s` still usable after move-by-value
+        assert_eq!(a, b);
+        assert_eq!(s, a);
+    }
+}