Aura/docs/specs/0041-source-ingestion-seam.md

Source ingestion seam — Design Spec

Date: 2026-06-15 Status: Draft — awaiting user spec review Authors: orchestrator + Claude

Goal

Replace the engine's single materialized-Vec ingestion input with a producer seam — a Source trait the k-way merge drives by peek/next — and prove it against a real, lazily-streamed data-server window in the same cycle, so the seam is designed against the real data source from the first line.

Harness::run today takes Vec<Vec<(Timestamp, Scalar)>>: one fully materialized stream per source. That is the only engine type encoding an eager-dataflow assumption. At realistic research scale (20 years, 3 tick streams, ~7.9e9 ticks) the eager layout is ~190 GB resident (24 B/tick) — non-residable. The merge loop itself never needs the whole stream: it only ever peeks the head timestamp of each live source and pops one record (harness.rs:237–257). So the eager Vec is gratuitous; a peek/next producer is the faithful seam, and a data-server-backed producer streams Arc<[T]> chunks lazily — resident footprint O(Σ node lookbacks + one chunk), flat across the horizon.

This is the foundational cycle of the World-II milestone: the orchestration families (#66 / #68 / #69) all bootstrap harnesses over data windows, so they inherit this seam. Ledger cycle 0011 (INDEX.md:530) already names today's eager materialization "a known, deliberate gap" whose zero-copy Arc<[T]> cross-sim sharing is "the target for the orchestration cycle that introduces [the axes]" — this milestone. This cycle closes that gap by making a streaming path exist; it does not delete the eager convenience path (still valid for bounded loads).

Contracts touched: C2 (bounded read-only windows), C3 (one merge at ingestion, ms→epoch-ns normalized there), C4 (cycle granularity / tie-by-source- index), C7 (SoA base columns), C12 (Arc<[T]> cross-sim sharing).

Architecture

Three layers, one new boundary type:

1. aura-engine gains a public Source trait and a VecSource adapter. Harness::run is re-typed from Vec<Vec<(Timestamp, Scalar)>> to Vec<Box<dyn Source>>. The merge loop's index arithmetic (streams[s][cursor[s]]) becomes sources[s].peek() / sources[s].next(). Every existing call site wraps its Vec<(Timestamp, Scalar)> in VecSource::new(..) — behaviour-preserving, byte-identical run output, no residency change. VecSource is the old behaviour, named.

2. aura-ingest gains M1FieldSource: a streaming Source that holds a data-server SymbolChunkIter<M1Parsed>, the current Arc<[M1Parsed]> chunk (an Arc clone of the cache's chunk — zero-copy), a cursor into it, a field selector, and a pre-decoded head record. It constructs each Scalar per-pull from the borrowed chunk (no owned-Vec materialization of the window) and refills via next_chunk() when the current chunk drains.

3. End-to-end: a gated integration test bootstraps the signal-quality sample harness, opens an M1FieldSource over a real [from, to] window, runs it through the seam to a RunReport, and asserts determinism (C1).

The peek(&self)/next(&mut self) split is what forces the pre-decoded head in the streaming source: answering peek without &mut is impossible over an underlying iterator whose refill (next_chunk) is &mut, so the head is decoded ahead of demand at construction and after every next. This is the standard peekable shape, determined by the trait signature (settled in the issue), not an open design choice.

SourceSpec (harness.rs:49) is unchanged: it stays the wiring metadata (scalar kind + target slots), declared on the FlatGraph. A Box<dyn Source> is the runtime producer bound to source index s at run time. The two were already distinct (the old Vec was the runtime half); this cycle only re-types the runtime half from an owned Vec to a producer.

Scope boundary

In scope: the trait, VecSource, the run re-type with every call site adapted, M1FieldSource (the streaming source), and one streaming end-to-end test.

Out of scope (deliberately, to keep the seam focused):

• Deleting the eager load_m1_window / close_stream path — it stays a valid convenience for bounded loads; the CLI keeps using it (wrapped in VecSource). The deliberate-eager gap is closed by a streaming path existing, not by removing the eager one.
• Migrating the CLI's real-data commands to streaming — follow-on.
• Tick sources (stream_tick_windowed) — the same shape applies (M1FieldSource generalizes), but the e2e and the SMA-cross sample consume M1 close; tick is not built here.
• Seed-as-input (#66) — the seed lives at the recording edge, outside this seam.

Concrete code shapes

User-facing program — a streaming real-window backtest

What a project author (or a World program) writes to backtest a strategy over a real data window. This is the acceptance criterion's empirical evidence: the audience's natural reach, shown.

// Bootstrap the harness (unchanged — topology is FlatGraph as today).
let mut h = Harness::bootstrap(sma_cross_signal_quality_graph())?;

// Open a streaming source over a real window. None = no archived file overlaps
// the window (data-server's file-level Option contract, propagated).
let close = M1FieldSource::open(&server, "AAPL.US", Some(from_ms), Some(to_ms), M1Field::Close)
    .expect("window overlaps real data");

// Drive the seam. The source streams Arc<[M1Parsed]> chunks lazily: resident
// footprint is one chunk + the engine's per-node lookback windows, NOT the
// whole window. A 20-year window streams in the same memory as a one-day one.
h.run(vec![Box::new(close)]);

let report = fold_recorded_into_report(/* sinks */);

The eager path still works unchanged for bounded loads (every existing call site, now wrapped):

let prices: Vec<(Timestamp, Scalar)> =
    load_m1_window(&server, "AAPL.US", Some(from_ms), Some(to_ms))?.close_stream();
h.run(vec![Box::new(VecSource::new(prices))]); // was: h.run(vec![prices]);

Delivered code — the Source trait (aura-engine)

/// The ingestion-boundary producer the k-way merge drives (C3). A source yields
/// timestamped scalars in ascending timestamp order (the C3 ingestion
/// precondition). Object-safe: the merge holds `Vec<Box<dyn Source>>`.
pub trait Source {
    /// The timestamp of the head record, without consuming it. `None` = the
    /// source is exhausted. Cheap and side-effect-free: it reads a pre-decoded
    /// head, never advances the underlying stream.
    fn peek(&self) -> Option<Timestamp>;

    /// Pop the head `(timestamp, scalar)` and advance. `None` = exhausted.
    /// After `next` returns `Some`, `peek` reflects the new head.
    fn next(&mut self) -> Option<(Timestamp, Scalar)>;
}

Delivered code — VecSource (aura-engine), behaviour-preserving

/// A `Source` over a fully materialized stream — the cycle-0011 eager shape,
/// named. Every pre-seam call site wraps its `Vec<(Timestamp, Scalar)>` in this,
/// so run output is byte-identical (C1) and residency is unchanged.
pub struct VecSource {
    stream: Vec<(Timestamp, Scalar)>,
    cursor: usize,
}

impl VecSource {
    pub fn new(stream: Vec<(Timestamp, Scalar)>) -> Self {
        Self { stream, cursor: 0 }
    }
}

impl Source for VecSource {
    fn peek(&self) -> Option<Timestamp> {
        self.stream.get(self.cursor).map(|&(t, _)| t)
    }
    fn next(&mut self) -> Option<(Timestamp, Scalar)> {
        let item = self.stream.get(self.cursor).copied()?; // (Timestamp, Scalar): Copy
        self.cursor += 1;
        Some(item)
    }
}

Harness::run — the merge loop re-type (before → after)

// BEFORE (harness.rs:220) — eager Vec input, index arithmetic.
pub fn run(&mut self, streams: Vec<Vec<(Timestamp, Scalar)>>) {
    assert_eq!(streams.len(), self.sources.len(), /* ... */);
    let mut cursor: Vec<usize> = vec![0; streams.len()];
    loop {
        let mut pick: Option<usize> = None;
        for (s, stream) in streams.iter().enumerate() {
            if cursor[s] < stream.len() {
                match pick {
                    None => pick = Some(s),
                    Some(p) => if stream[cursor[s]].0 < streams[p][cursor[p]].0 { pick = Some(s); }
                }
            }
        }
        let s = match pick { Some(s) => s, None => break };
        let (ts, value) = streams[s][cursor[s]];
        cursor[s] += 1;
        /* ... forward into target slots, eval in topo order ... unchanged ... */
    }
}

// AFTER — producer seam, peek/next. The pick scan, tie-by-source-index, and the
// entire forward+eval body are preserved; only the head access changes.
pub fn run(&mut self, mut sources: Vec<Box<dyn Source>>) {
    assert_eq!(sources.len(), self.sources.len(), /* ... */);
    loop {
        // pick the live source head with the smallest (timestamp, source index).
        // strictly-`<` replace + source-order scan preserves C4 tie-breaking.
        let mut pick: Option<(usize, Timestamp)> = None;
        for (s, src) in sources.iter().enumerate() {
            if let Some(ts) = src.peek() {
                match pick {
                    None => pick = Some((s, ts)),
                    Some((_, best)) if ts < best => pick = Some((s, ts)),
                    _ => {}
                }
            }
        }
        let s = match pick { Some((s, _)) => s, None => break };
        let (ts, value) = sources[s].next().expect("peeked Some ⇒ next is Some");
        /* ... forward into target slots, eval in topo order ... byte-identical ... */
    }
}

Note the borrow shape: self.sources (the SourceSpec wiring) is read inside the forward step while sources (the producers) is the separate run argument, so the existing destructuring of self into disjoint field borrows is unaffected.

Delivered code — M1FieldSource (aura-ingest), the streaming source

/// Which base column of an M1 bar a source streams (C7: a composite window is a
/// bundle of base columns; one `Source` per consumed field).
pub enum M1Field { Open, High, Low, Close, Spread, Volume }

/// A streaming `Source` over a data-server M1 window. Holds one Arc chunk
/// (zero-copy clone of the cache's chunk) and a cursor; constructs each `Scalar`
/// per-pull; refills via `next_chunk()` when the chunk drains. Resident footprint
/// is O(one chunk), independent of window length.
pub struct M1FieldSource {
    iter: SymbolChunkIter<M1Parsed>,    // data-server's lazy chunk producer
    chunk: Option<Arc<[M1Parsed]>>,     // current chunk, borrowed (Arc refcount, not a data copy)
    pos: usize,                         // cursor within `chunk`
    field: M1Field,                     // which column this source projects
    head: Option<(Timestamp, Scalar)>,  // pre-decoded head ⇒ peek(&self) is cheap
}

impl M1FieldSource {
    /// Open over `[from_ms, to_ms]` (inclusive Unix-ms, data-server's contract).
    /// `None` when no archived file overlaps the window (an unknown symbol or a
    /// far-future window) — propagating `stream_m1_windowed`'s file-level Option.
    /// A window that overlaps a file but holds zero matching bars yields a source
    /// whose first `peek` is `None` (immediately exhausted), not `None` here.
    pub fn open(
        server: &Arc<DataServer>, symbol: &str,
        from_ms: Option<i64>, to_ms: Option<i64>, field: M1Field,
    ) -> Option<Self> {
        let iter = server.stream_m1_windowed(symbol, from_ms, to_ms)?;
        let mut s = Self { iter, chunk: None, pos: 0, field, head: None };
        s.advance(); // pre-decode the first head
        Some(s)
    }

    /// Project the bar at the cursor into (normalized ts, scalar) for `field`.
    fn decode(&self, bar: &M1Parsed) -> (Timestamp, Scalar) {
        let value = match self.field {
            M1Field::Open => Scalar::F64(bar.open),
            M1Field::High => Scalar::F64(bar.high),
            M1Field::Low => Scalar::F64(bar.low),
            M1Field::Close => Scalar::F64(bar.close),
            M1Field::Spread => Scalar::F64(bar.spread),
            M1Field::Volume => Scalar::I64(bar.volume),
        };
        (unix_ms_to_epoch_ns(bar.time_ms), value) // ms → epoch-ns at the seam (C3)
    }

    /// Decode the head at the cursor, refilling chunks as needed. Sets
    /// `self.head = None` at exhaustion.
    fn advance(&mut self) {
        loop {
            if self.chunk.is_none() {
                self.chunk = self.iter.next_chunk(); // &mut refill; None = window exhausted
                self.pos = 0;
                if self.chunk.is_none() { self.head = None; return; }
            }
            let chunk = self.chunk.as_ref().unwrap();
            if self.pos < chunk.len() {
                self.head = Some(self.decode(&chunk[self.pos]));
                return;
            }
            self.chunk = None; // current chunk drained — loop to refill
        }
    }

    /// Records resident in this source right now: the current chunk's length (or
    /// 0 at exhaustion). The residency probe — bounded by one chunk length by
    /// construction (the struct holds at most one `Arc` chunk and no accumulating
    /// field), so it can never grow with window length. Sampled by the residency
    /// test to make the O(one chunk) claim a measured quantity, not prose.
    pub fn resident_records(&self) -> usize {
        self.chunk.as_ref().map_or(0, |c| c.len())
    }
}

impl Source for M1FieldSource {
    fn peek(&self) -> Option<Timestamp> {
        self.head.map(|(t, _)| t)
    }
    fn next(&mut self) -> Option<(Timestamp, Scalar)> {
        let item = self.head?;
        self.pos += 1;
        self.advance(); // pre-decode the next head (may refill the chunk)
        Some(item)
    }
}

N-field generalization (shape the trait permits; not all-tested here)

Two M1FieldSources over the same (symbol, window) get the same Arc chunks from the data-server cache (the SymbolGuard keeps the symbol loaded; the cache shares Arc<[T]>), so they stream different fields sharing the chunk's ts axis zero-copy — C12's cross-source Arc<[T]> sharing:

let close  = M1FieldSource::open(&server, sym, f, t, M1Field::Close)?;
let volume = M1FieldSource::open(&server, sym, f, t, M1Field::Volume)?;
h.run(vec![Box::new(close), Box::new(volume)]); // co-scheduled by timestamp (C4)

The e2e headline uses the close source alone; a focused test asserts two sources over one window both stream (the sharing property), but the full OHLCV-as-N- sources harness is not built in this cycle.

Components

Component	Crate	Change
Source trait	aura-engine	new — peek(&self), next(&mut); object-safe; re-exported from the crate root
VecSource	aura-engine	new — behaviour-preserving adapter over Vec<(Timestamp, Scalar)>
Harness::run	aura-engine	re-typed — Vec<Box<dyn Source>>; merge loop drives peek/next; forward+eval body byte-identical
~53 run call sites	aura-engine, aura-cli, aura-ingest	adapted — wrap streams in VecSource::new(..); mostly tests
M1Field, M1FieldSource	aura-ingest	new — streaming M1 source, per-pull decode, lazy chunk refill, ms→ns normalization
streaming e2e test	aura-ingest	new — real-window backtest through the seam to a RunReport
load_m1_window, close_stream, transpose_m1	aura-ingest	kept — eager convenience for bounded loads; unchanged

Data flow

data-server files ──stream_m1_windowed(sym, from_ms, to_ms)──▶ SymbolChunkIter<M1Parsed>
                                                                    │  next_chunk() : Arc<[M1Parsed]>  (lazy, N+2 prefetch)
                                                                    ▼
M1FieldSource { chunk: Arc<[M1Parsed]>, pos, field, head } ──decode(bar)──▶ (epoch-ns ts, Scalar)
                                                                    │  peek() / next()
                                                                    ▼
Harness::run(Vec<Box<dyn Source>>) ──k-way merge by (peek ts, source idx)──▶ one chronological cycle stream
                                                                    │  (forward into target slots → eval in topo order — unchanged)
                                                                    ▼
                                                             recorded sinks → RunReport

The only resident data at any instant: each source's current Arc chunk (≤1 chunk) + one pre-decoded head, plus the engine's per-node lookback ring buffers (already O(Σ lookbacks), unchanged). Nothing scales with window length.

Error handling

• Source open failure is an Option, not a Result: M1FieldSource::open returns None exactly when stream_m1_windowed does — no archived file overlaps the window (unknown symbol / far-future window). This mirrors the existing load_m1_window file-level Option contract (lib.rs:101–109).
• Empty-but-overlapping window (a file overlaps but holds zero matching bars): open returns Some, and the source's first peek is None — immediately exhausted. run over it drives zero cycles. This preserves the existing distinction "data source present" vs "nothing to read from".
• Mid-stream: next/peek never error — exhaustion is None. next_chunk yields None at window end; advance maps that to head = None.
• run's length assertion (one source per SourceSpec) is unchanged.
• The ascending-timestamp ingestion precondition (C3) is unchanged: data-server files are chronological, and decode preserves order within and across chunks.

Testing strategy

1. Behaviour-preserving (the load-bearing guarantee). The full cargo test --workspace suite stays green and byte-identical after the re-type — every call site wrapped in VecSource. The existing C1-determinism assertions (e.g. real_bars.rs:118–120, blueprint.rs sweep equality) are the evidence; no test output changes.
2. VecSource unit tests (aura-engine): peek is non-consuming and idempotent; peek then next agree; exhaustion yields None; an empty stream peeks None.
3. decode projection unit tests (aura-ingest, hermetic — pure decode on hand-made M1Parsed, no data-server files): each M1Field projects to the right Scalar kind/value with ts normalized ms→ns. decode is a pure function of (bar, field), so it is fully testable without an iterator or files. (The peek/next/refill walk over a real iterator is exercised by the gated e2e below, which is where a real SymbolChunkIter exists.)
4. Streaming e2e + residency proof (aura-ingest, gated — skips where the local Pepperstone directory is absent, like real_bars.rs): bootstrap the signal-quality sample harness, M1FieldSource::open over a real window spanning many data-server chunks, h.run through the seam, fold to a RunReport; assert total_pips.is_finite() and C1 (two runs bit-identical JSON). Residency predicate — the measurable O(one chunk) claim: sampling resident_records() at every pull across the whole run, its maximum is ≤ one data-server chunk — the CHUNK_SIZE bound (1024 records; data_server's documented chunk size, asserted against the constant, not a magic literal). This per-pull bound is independent of window length: it holds at every pull whether the window spans a handful of chunks or millions, which is the O(one chunk)-not-O(window) content. The bound is correct on real chunks regardless of boundary-chunk lengths (filter_chunk only ever yields a chunk ≤ CHUNK_SIZE). A source that accumulated chunks would blow past the one-chunk bound on a long window; a source that accumulated decoded state is excluded structurally by §5. (Deliberately not an "equal peak across two windows" assertion — boundary/file-final chunks are short, so per-window maxima legitimately differ; the load-bearing invariant is the per-pull one-chunk ceiling, which is what bounds residency.)
5. Structural no-accumulation guarantee (always-on, not gated): M1FieldSource holds exactly { iter, one Option<Arc<[M1Parsed]>>, cursor, field, one head } — no Vec<Arc<..>> or other field that grows with the stream — so resident_records() is bounded by one chunk length by construction. This is the always-on half of the residency claim (verifiable from the struct definition + clippy); the gated item 4 is the dynamic-invariance half.
6. Sharing property (gated): two M1FieldSources (close + volume) over one window both stream to exhaustion in a single run — the N-sources-share-the- window shape.

Acceptance criteria

• Source (peek + next, object-safe) and VecSource exist in aura-engine; Harness::run takes Vec<Box<dyn Source>>.
• Every run call site is adapted; cargo test --workspace is green and run output is byte-identical to pre-seam (behaviour-preserving).
• M1FieldSource streams a data-server M1 window through the seam: per-pull Scalar decode from a borrowed Arc chunk, lazy next_chunk refill, ms→epoch-ns normalized at the seam (C3).
• A real short-window backtest runs end-to-end through the seam to a RunReport, deterministically (C1).
• Residency (measured predicate): across a full run over a window spanning many chunks, peak M1FieldSource::resident_records() is ≤ one data-server chunk (CHUNK_SIZE = 1024) — a per-pull ceiling independent of window length — so the source holds O(one chunk), NOT O(window length), measured on real multi-chunk data (Testing strategy §4). The struct carries no accumulating field, so this bound holds by construction (§5). The engine's per-node lookback ring buffers are unchanged — already O(Σ lookbacks), pinned by the existing suite — so total resident memory is flat across the horizon.
• cargo clippy --workspace --all-targets -- -D warnings clean; cargo doc --workspace --no-deps clean.

Surfaced boundary frictions (recorded, per the issue's friction-first intent)

• Transpose granularity. The streaming source does per-pull field decode from the borrowed chunk, not transpose_m1's per-window owned-SoA transpose. The eager M1Columns/transpose_m1 path is a different shape, kept for bounded loads; the two now coexist (a follow-on may unify them).
• Arc borrow lifetime. Each source holds an Arc<[M1Parsed]> clone; across disjoint sims (C1/C12) the data-server cache shares the same chunks zero-copy. The SymbolGuard inside SymbolChunkIter keeps the symbol loaded for the source's lifetime.
• Data-server cache file retention. The aura-side source is O(one chunk); the data-server cache retains loaded files (with N+2 prefetch). Whether files behind the cursor are evicted is the data-server's contract, owned by that lib, not this cycle — the residency claim here is scoped to aura's source footprint. Flagged for the milestone, not resolved here.
• Error channel asymmetry. Option at open (file-level), None-on- exhaustion mid-stream — no Result anywhere, matching stream_m1_windowed.
• peek(&self) vs &mut refill. Resolved by the pre-decoded head; the head decode happens in advance, called from open and next.