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

Source ingestion seam — Implementation Plan

Parent spec: docs/specs/0041-source-ingestion-seam.md

For agentic workers: REQUIRED SUB-SKILL: use the implement skill to run this plan. Steps use - [ ] checkboxes for tracking.

Goal: Re-type the engine's ingestion input from a materialized Vec<Vec<(Timestamp, Scalar)>> into a Source producer trait the k-way merge drives by peek/next, and prove it against a real, lazily-streamed data-server window — closing the cycle-0011 eager-materialization gap.

Architecture: aura-engine gains a Source trait + a VecSource adapter; Harness::run is re-typed to Vec<Box<dyn Source>> with the merge loop driven by peek/next, and every existing call site is wrapped in VecSource (behaviour-preserving, byte-identical). aura-ingest gains a streaming M1FieldSource over a data-server M1 window (per-pull decode from a borrowed Arc chunk, lazy next_chunk refill, ms→epoch-ns at the seam), proven by a gated end-to-end backtest plus a measured residency predicate.

Tech Stack: aura-engine (harness.rs, lib.rs), aura-ingest (lib.rs, tests/), data-server (stream_m1_windowed, SymbolChunkIter, next_chunk, loader::CHUNK_SIZE, M1Parsed).

---

Files this plan creates or modifies:

• Modify: crates/aura-engine/src/harness.rs — add Source trait + VecSource (after SourceSpec, ~line 52); re-type Harness::run (line 220) + merge loop (220-258); re-type test helper ohlcv_streams() (line 900); wrap 31 call sites.
• Modify: crates/aura-engine/src/lib.rs:25-30,47 — refresh module doc; add Source, VecSource to the crate-root re-export.
• Modify: crates/aura-engine/src/blueprint.rs — wrap 10 call sites.
• Modify: crates/aura-engine/src/sweep.rs:263 — wrap 1 call site.
• Modify: crates/aura-engine/src/report.rs:223 — wrap 1 call site.
• Modify: crates/aura-cli/src/main.rs — wrap 4 call sites.
• Modify: crates/aura-ingest/src/lib.rs — add M1Field + decode + M1FieldSource (near load_m1_window, ~line 110); add data_server::SymbolChunkIter import.
• Modify: crates/aura-ingest/tests/real_bars.rs:72 — wrap 1 call site.
• Create: crates/aura-ingest/tests/streaming_seam.rs — gated streaming e2e + residency-drive + sharing tests.
• Test: harness.rs #[cfg(test)] mod tests — VecSource unit tests.
• Test: aura-ingest/src/lib.rs #[cfg(test)] mod tests — decode unit tests.

---

Task 1: Source trait + VecSource adapter (aura-engine)

Files:

• Modify: crates/aura-engine/src/harness.rs (after SourceSpec, ~line 52)
• Modify: crates/aura-engine/src/lib.rs:25-30,47
• Test: crates/aura-engine/src/harness.rs (#[cfg(test)] mod tests)

This task adds the new types only; Harness::run is NOT re-typed here (that is Task 2). Source/VecSource compile and their tests pass standalone.

• Step 1: Add the Source trait + VecSource to harness.rs

Insert immediately after the SourceSpec struct (after its closing } at harness.rs:52, before the FlatGraph doc comment at line 54):

/// 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 advancing 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)>;
}

/// 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()?;
        self.cursor += 1;
        Some(item)
    }
}

• Step 2: Add VecSource unit tests

Append inside the existing #[cfg(test)] mod tests { ... } block in harness.rs (the module already has use super::*;):

#[test]
fn vec_source_peek_is_non_consuming_and_idempotent() {
    let mut s = VecSource::new(vec![(Timestamp(1), Scalar::F64(10.0)), (Timestamp(2), Scalar::F64(20.0))]);
    // peek twice: same head, no advance.
    assert_eq!(s.peek(), Some(Timestamp(1)));
    assert_eq!(s.peek(), Some(Timestamp(1)));
    // next pops it; peek now reflects the new head.
    assert_eq!(Source::next(&mut s), Some((Timestamp(1), Scalar::F64(10.0))));
    assert_eq!(s.peek(), Some(Timestamp(2)));
}

#[test]
fn vec_source_exhaustion_yields_none() {
    let mut s = VecSource::new(vec![(Timestamp(5), Scalar::I64(7))]);
    assert_eq!(Source::next(&mut s), Some((Timestamp(5), Scalar::I64(7))));
    assert_eq!(s.peek(), None);
    assert_eq!(Source::next(&mut s), None);
}

#[test]
fn vec_source_empty_peeks_none() {
    let mut s = VecSource::new(vec![]);
    assert_eq!(s.peek(), None);
    assert_eq!(Source::next(&mut s), None);
}

(Source::next(&mut s) is spelled fully-qualified so it never collides with Iterator::next in scope.)

• Step 3: Re-export the new types and refresh the module doc

In crates/aura-engine/src/lib.rs, change the re-export at line 47 from:

pub use harness::{BootstrapError, Edge, FlatGraph, Harness, SourceSpec, Target};

to:

pub use harness::{BootstrapError, Edge, FlatGraph, Harness, Source, SourceSpec, Target, VecSource};

And in the module doc, change the "Still to come" line at lines 25-26 from:

//! Still to come (subsequent cycles): the `Source` trait + data-server ingestion
//! and source-native time normalization (C3/C11), the broker-independent

to:

//! Still to come (subsequent cycles): the broker-independent

• Step 4: Build and run the VecSource tests

Run: cargo test -p aura-engine vec_source Expected: PASS (3 tests: vec_source_peek_is_non_consuming_and_idempotent, vec_source_exhaustion_yields_none, vec_source_empty_peeks_none).

• Step 5: Lint gate

Run: cargo clippy -p aura-engine --all-targets -- -D warnings Expected: clean (no warnings; the pub re-exported Source/VecSource are not dead code).

---

Task 2: Re-type Harness::run + thread all callers (atomic compile unit)

Files:

• Modify: crates/aura-engine/src/harness.rs:220-258 (signature + merge loop), :900 (ohlcv_streams helper), 31 call sites
• Modify: crates/aura-engine/src/blueprint.rs (10 sites)
• Modify: crates/aura-engine/src/sweep.rs:263 (1 site)
• Modify: crates/aura-engine/src/report.rs:223 (1 site)
• Modify: crates/aura-cli/src/main.rs (4 sites)
• Modify: crates/aura-ingest/tests/real_bars.rs:72 (1 site)

Why one task: re-typing run's signature makes the whole workspace fail to compile until every one of the 53 callers is threaded. The signature change and all callers must land together; the build gate at the end is unsatisfiable otherwise (self-review item 7). The 4 h.run(ohlcv_streams()) sites are threaded by re-typing the helper, not the call site.

The mechanical wrap rule (applies to every vec![...] call-site literal): each top-level element X of the vec![...] passed to .run( is replaced by Box::new(VecSource::new(X)). The expected parameter type Vec<Box<dyn Source>> drives the unsizing coercion, so no as Box<dyn Source> is needed. Example: h.run(vec![prices]) → h.run(vec![Box::new(VecSource::new(prices))]); h.run(vec![s0, s1]) → h.run(vec![Box::new(VecSource::new(s0)), Box::new(VecSource::new(s1))]).

• Step 1: Re-type the run signature + merge loop in harness.rs

Replace the current signature + pick scan (harness.rs:220-257). Current:

    pub fn run(&mut self, streams: Vec<Vec<(Timestamp, Scalar)>>) {
        assert_eq!(
            streams.len(),
            self.sources.len(),
            "run: one stream per source required (got {} streams for {} sources)",
            streams.len(),
            self.sources.len()
        );

        // disjoint field borrows so the topo walk can read topo/out_edges/sources
        // while mutating nodes
        let Harness { nodes, topo, out_edges, sources } = self;

        let mut cursor: Vec<usize> = vec![0; streams.len()];
        let mut cycle_id: u64 = 0;
        let mut scratch: Vec<Scalar> = Vec::new();

        loop {
            // pick the live source head with the smallest (timestamp, source index)
            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, // all streams exhausted
            };
            let (ts, value) = streams[s][cursor[s]];
            cursor[s] += 1;
            cycle_id += 1;

Replace with (sources arg is mut; the forward+eval body below line 257 is UNCHANGED):

    pub fn run(&mut self, mut sources_in: Vec<Box<dyn Source>>) {
        assert_eq!(
            sources_in.len(),
            self.sources.len(),
            "run: one source per SourceSpec required (got {} sources for {} specs)",
            sources_in.len(),
            self.sources.len()
        );

        // disjoint field borrows so the topo walk can read topo/out_edges/sources
        // while mutating nodes
        let Harness { nodes, topo, out_edges, sources } = self;

        let mut cycle_id: u64 = 0;
        let mut scratch: Vec<Scalar> = Vec::new();

        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_in.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, // all sources exhausted
            };
            // `&mut *sources_in[s]`: deref the Box to `dyn Source` then re-borrow,
            // so UFCS resolves Self = dyn Source (Box<dyn Source> does not itself
            // impl Source). Fully-qualified to never collide with Iterator::next.
            let (ts, value) =
                Source::next(&mut *sources_in[s]).expect("peeked Some ⇒ next is Some");
            cycle_id += 1;

(The local binding is named sources_in to avoid shadowing the sources destructured from self — the SourceSpec wiring used in the forward step at the unchanged line for t in sources[s].targets.iter().)

• Step 2: Verify the forward/eval body still references sources (the wiring)

Confirm the unchanged forwarding line (was harness.rs:261) still reads:

            for t in sources[s].targets.iter() {

sources here is the Vec<SourceSpec> destructured from self — unchanged. No edit; this step is a read-check that the rename in Step 1 did not collide.

• Step 3: Re-type the ohlcv_streams() test helper in harness.rs:900

Replace (lines 899-908):

    /// Build five timestamp-aligned f64 sources feeding Ohlcv's five barrier slots.
    fn ohlcv_streams() -> Vec<Vec<(Timestamp, Scalar)>> {
        vec![
            f64_stream(&[(1, 10.0), (2, 20.0)]), // open
            f64_stream(&[(1, 15.0), (2, 25.0)]), // high
            f64_stream(&[(1, 8.0), (2, 19.0)]),  // low
            f64_stream(&[(1, 12.0), (2, 22.0)]), // close
            f64_stream(&[(1, 100.0), (2, 200.0)]), // volume
        ]
    }

with (returns boxed sources, so the four h.run(ohlcv_streams()) callers need no change):

    /// Build five timestamp-aligned f64 sources feeding Ohlcv's five barrier slots.
    fn ohlcv_streams() -> Vec<Box<dyn Source>> {
        vec![
            Box::new(VecSource::new(f64_stream(&[(1, 10.0), (2, 20.0)]))), // open
            Box::new(VecSource::new(f64_stream(&[(1, 15.0), (2, 25.0)]))), // high
            Box::new(VecSource::new(f64_stream(&[(1, 8.0), (2, 19.0)]))),  // low
            Box::new(VecSource::new(f64_stream(&[(1, 12.0), (2, 22.0)]))), // close
            Box::new(VecSource::new(f64_stream(&[(1, 100.0), (2, 200.0)]))), // volume
        ]
    }

• Step 4: Wrap the 31 remaining harness.rs call sites

Apply the wrap rule to each site below (the 4 h.run(ohlcv_streams()) at :948/:1000/:1013/:1043 are NOT in this list — they are handled by Step 3). Each vec![...] element becomes Box::new(VecSource::new(<element>)):

:594  h.run(vec![f64_stream(&[(1,1.0),(2,2.0),(3,3.0),(4,4.0),(5,5.0)])]);
:642  h.run(vec![prices.clone()]);            :657  h2.run(vec![prices]);
:689  h.run(vec![s0.clone(), s1.clone()]);    :704  h2.run(vec![s0, s1]);
:735  h.run(vec![s0.clone(), s1.clone()]);    :748  h2.run(vec![s0, s1]);
:789  h.run(vec![prices.clone()]);            :804  h2.run(vec![prices]);
:834  h.run(vec![s0, s1, s2]);
:1122 h.run(vec![f64_stream(&[(1,10.0),(2,12.0),(3,14.0),(4,16.0),(5,18.0)])]);
:1172 h.run(vec![ f64_stream(..)×5 ]);         (wrap each of the 5 elements)
:1219 h.run(vec![ <4 raw inner-vec literals> ]); (wrap each of the 4 — see Step 5)
:1257 h.run(vec![f64_stream(&[(1,10.0),(2,20.0),(3,30.0)])]);
:1292 a.run(vec![prices.clone()]);            :1297 b.run(vec![prices]);
:1327 h.run(vec![s0, s1]);                     :1362 h.run(vec![s0, s1]);
:1447 h.run(vec![prices.clone()]);            :1466 h2.run(vec![prices]);
:1512 h.run(vec![prices.clone()]);            :1535 h2.run(vec![prices]);
:1579 h.run(vec![a.clone(), b.clone()]);       :1600 h2.run(vec![a, b]);
:1640 h.run(vec![prices.clone()]);            :1651 h2.run(vec![prices]);
:1701 h.run(vec![prices.clone()]);            :1728 h2.run(vec![prices]);
:1788 h.run(vec![prices.clone(), counts.clone()]); :1817 h2.run(vec![prices, counts]);
:1861 h.run(vec![f64_stream(&[...])]);          (multi-line; wrap the single element)
:1926 h.run(vec![f64_stream(&[...])]);          (multi-line; wrap the single element)

(Line numbers drift as edits are applied; match on the call expression, not the absolute line. After this step, git grep -n 'run(vec!' crates/aura-engine/src/harness.rs should return zero un-wrapped vec![ — every element is now Box::new(VecSource::new(...)).)

• Step 5: Wrap the raw inner-vec literal at harness.rs:1219

Replace:

        h.run(vec![
            vec![(Timestamp(1), Scalar::I64(7))],
            vec![(Timestamp(2), Scalar::F64(1.5))],
            vec![(Timestamp(3), Scalar::Bool(true))],
            vec![(Timestamp(4), Scalar::Ts(Timestamp(99)))],
        ]);

with:

        h.run(vec![
            Box::new(VecSource::new(vec![(Timestamp(1), Scalar::I64(7))])),
            Box::new(VecSource::new(vec![(Timestamp(2), Scalar::F64(1.5))])),
            Box::new(VecSource::new(vec![(Timestamp(3), Scalar::Bool(true))])),
            Box::new(VecSource::new(vec![(Timestamp(4), Scalar::Ts(Timestamp(99)))])),
        ]);

• Step 6: Wrap the 10 blueprint.rs call sites

Apply the wrap rule to each (match on the expression):

:1023 named.run(vec![synthetic_prices()]);     :1031 positional.run(vec![synthetic_prices()]);
:1742 flat.run(vec![prices.clone()]);          :1749 composed.run(vec![prices]);
:1812 h.run(vec![prices]);                      :1829 a.run(vec![prices.clone()]);
:1835 b.run(vec![prices]);                      :1873 a.run(vec![prices.clone()]);
:1877 b.run(vec![prices]);                      :2395 h.run(vec![prices]);

Each → <recv>.run(vec![Box::new(VecSource::new(<element>))]);.

• Step 7: Wrap the sweep.rs and report.rs call sites

crates/aura-engine/src/sweep.rs:263: h.run(vec![synthetic_prices()]); → h.run(vec![Box::new(VecSource::new(synthetic_prices()))]);

crates/aura-engine/src/report.rs:223 (multi-line f64_stream element): wrap the single element → h.run(vec![Box::new(VecSource::new(f64_stream(&[...])))]);

• Step 8: Wrap the 4 aura-cli/src/main.rs call sites

:140  h.run(vec![prices]);          :283  h.run(vec![prices]);
:462  h.run(vec![prices]);          :527  h.run(vec![showcase_prices()]);

Each → h.run(vec![Box::new(VecSource::new(<element>))]);. main.rs already imports from aura_engine; ensure VecSource is in the import list (the file imports Harness etc. from aura_engine::{...} — add VecSource).

• Step 9: Wrap the real_bars.rs call site

crates/aura-ingest/tests/real_bars.rs:72: h.run(vec![prices]); → h.run(vec![Box::new(VecSource::new(prices))]); Add VecSource to the use aura_engine::{...} import list at real_bars.rs:11.

• Step 10: Workspace build gate

Run: cargo build --workspace --tests Expected: 0 errors (every one of the 53 callers threaded; the only fn run in the workspace is Harness::run, so there is no method-resolution collision).

• Step 11: Behaviour-preserving test gate

Run: cargo test --workspace Expected: all green — byte-identical to pre-seam. The two-run C1 determinism pairs (real_bars.rs r1/r2 to_json, blueprint.rs sweep-equality, harness.rs h/h2 pairs) stay green; no test output changes.

• Step 12: Lint gate

Run: cargo clippy --workspace --all-targets -- -D warnings Expected: clean.

---

Task 3: M1Field + M1FieldSource streaming source (aura-ingest)

Files:

• Modify: crates/aura-ingest/src/lib.rs (imports at 15-18; new items near load_m1_window at ~110)
• Test: crates/aura-ingest/src/lib.rs (#[cfg(test)] mod tests)

decode is a free function of (field, bar) (the spec calls it "a pure function of (bar, field)"), so the hermetic decode test needs no iterator.

• Step 0: Move aura-engine to [dependencies]

M1FieldSource lives in src/lib.rs and implements aura_engine::Source, so aura-engine must be a regular dependency — it is currently [dev-dependencies] only (Cargo.toml:17). In crates/aura-ingest/Cargo.toml, add under [dependencies] (after the data-server line):

aura-engine = { path = "../aura-engine" }

and REMOVE the now-duplicate aura-engine = { path = "../aura-engine" } line from [dev-dependencies] (cargo warns on a dependency present in both). Leave aura-std in [dev-dependencies] — only the tests use it. Regular deps are visible to tests, so real_bars.rs still resolves aura_engine.

Run: cargo tree -p aura-ingest -e normal -i aura-engine 2>&1 | head -2 Expected: aura-engine now appears as a normal (non-dev) dependency.

• Step 1: Add the SymbolChunkIter import

In crates/aura-ingest/src/lib.rs, change line 17:

use data_server::DataServer;

to:

use data_server::{DataServer, SymbolChunkIter};

• Step 2: Add M1Field, decode, and M1FieldSource

Insert after load_m1_window (after its closing } at lib.rs:123):

/// 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).
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum M1Field {
    Open,
    High,
    Low,
    Close,
    Spread,
    Volume,
}

/// Project one M1 bar into `(normalized ts, scalar)` for `field`. Pure: a
/// function of `(field, bar)` only — no iterator, no clock — so it is testable
/// in isolation. Time is normalized ms→epoch-ns at this one seam (C3).
fn decode(field: M1Field, bar: &M1Parsed) -> (Timestamp, Scalar) {
    let value = match 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)
}

/// A streaming [`Source`](aura_engine::Source) over a data-server M1 window. Holds
/// at most one `Arc` chunk (a 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>,
    chunk: Option<Arc<[M1Parsed]>>,
    pos: usize,
    field: M1Field,
    head: Option<(Timestamp, Scalar)>,
}

impl M1FieldSource {
    /// Open over `[from_ms, to_ms]` (inclusive Unix-ms, data-server's contract).
    /// `None` when no archived file overlaps the window (unknown symbol /
    /// 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();
        Some(s)
    }

    /// 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();
                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(decode(self.field, &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 source holds at most one `Arc` chunk and no accumulating
    /// field), so it can never grow with window length.
    pub fn resident_records(&self) -> usize {
        self.chunk.as_ref().map_or(0, |c| c.len())
    }
}

impl aura_engine::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();
        Some(item)
    }
}

(aura-engine is made a regular dependency in Step 0, so aura_engine::Source resolves from src/lib.rs.)

• Step 4: Add hermetic decode unit tests

Append inside the existing #[cfg(test)] mod tests block in aura-ingest/src/lib.rs (it already has use super::*; and a full_bar helper at lib.rs:142):

#[test]
fn decode_projects_each_field_to_its_scalar_kind() {
    // full_bar: open 1.0, high 2.0, low 0.5, close 1.5, spread 0.1, volume 100.
    let bar = full_bar(1_000);
    assert_eq!(decode(M1Field::Open, &bar),   (Timestamp(1_000_000_000), Scalar::F64(1.0)));
    assert_eq!(decode(M1Field::High, &bar),   (Timestamp(1_000_000_000), Scalar::F64(2.0)));
    assert_eq!(decode(M1Field::Low, &bar),    (Timestamp(1_000_000_000), Scalar::F64(0.5)));
    assert_eq!(decode(M1Field::Close, &bar),  (Timestamp(1_000_000_000), Scalar::F64(1.5)));
    assert_eq!(decode(M1Field::Spread, &bar), (Timestamp(1_000_000_000), Scalar::F64(0.1)));
    // volume is the one i64 column.
    assert_eq!(decode(M1Field::Volume, &bar), (Timestamp(1_000_000_000), Scalar::I64(100)));
}

• Step 5: Build + run the decode tests + lint

Run: cargo test -p aura-ingest decode_projects Expected: PASS (decode_projects_each_field_to_its_scalar_kind).

Run: cargo clippy -p aura-ingest --all-targets -- -D warnings Expected: clean.

---

Task 4: Gated streaming e2e + residency proof (aura-ingest)

Files:

• Create: crates/aura-ingest/tests/streaming_seam.rs

Mirrors the gated pattern of real_bars.rs (skip where local data is absent). Three tests: the streaming backtest e2e, the residency-drive predicate, and the N-field sharing property.

• Step 1: Write the gated streaming test file

Create crates/aura-ingest/tests/streaming_seam.rs:

//! Gated integration test for the Source ingestion seam: a real data-server M1
//! close stream driven LAZILY through the signal-quality sample harness via
//! `M1FieldSource`, plus the residency predicate (peak resident records ≤ one
//! chunk, independent of window length) and the N-field sharing property. Skips
//! where the local Pepperstone data directory is absent.

use std::sync::mpsc;
use std::sync::Arc;

use aura_core::{Firing, NodeSchema, PortSpec, Scalar, ScalarKind, Timestamp};
use aura_engine::{
    f64_field, summarize, Edge, FlatGraph, Harness, RunManifest, RunReport, Source, SourceSpec,
    Target, VecSource,
};
use aura_ingest::{M1Field, M1FieldSource};
use aura_std::{Exposure, Recorder, SimBroker, Sma, Sub};
use data_server::loader::CHUNK_SIZE;
use data_server::{DataServer, DEFAULT_DATA_PATH};

const SYMBOL: &str = "AAPL.US";
// 2006-08 in inclusive Unix-ms (a full month of M1 ⇒ many > CHUNK_SIZE bars).
const FROM_MS: i64 = 1_154_390_400_000;
const TO_MS: i64 = 1_157_068_799_999;

/// Bootstrap the cycle-0007 two-sink signal-quality harness and run it on a
/// single boxed source, folding the recorded equity + exposure into a RunReport.
fn run_sample(window: (Timestamp, Timestamp), source: Box<dyn Source>) -> RunReport {
    let (tx_eq, rx_eq) = mpsc::channel();
    let (tx_ex, rx_ex) = mpsc::channel();
    let f64_recorder_sig = || NodeSchema {
        inputs: vec![PortSpec { kind: ScalarKind::F64, firing: Firing::Any, name: "in".into() }],
        output: vec![],
        params: vec![],
    };
    let mut h = Harness::bootstrap(FlatGraph {
        nodes: vec![
            Box::new(Sma::new(2)),
            Box::new(Sma::new(4)),
            Box::new(Sub::new()),
            Box::new(Exposure::new(0.5)),
            Box::new(SimBroker::new(0.0001)),
            Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx_eq)),
            Box::new(Recorder::new(&[ScalarKind::F64], Firing::Any, tx_ex)),
        ],
        signatures: vec![
            Sma::builder().schema().clone(),
            Sma::builder().schema().clone(),
            Sub::builder().schema().clone(),
            Exposure::builder().schema().clone(),
            SimBroker::builder(0.0001).schema().clone(),
            f64_recorder_sig(),
            f64_recorder_sig(),
        ],
        sources: vec![SourceSpec {
            kind: ScalarKind::F64,
            targets: vec![
                Target { node: 0, slot: 0 },
                Target { node: 1, slot: 0 },
                Target { node: 4, slot: 1 },
            ],
        }],
        edges: vec![
            Edge { from: 0, to: 2, slot: 0, from_field: 0 },
            Edge { from: 1, to: 2, slot: 1, from_field: 0 },
            Edge { from: 2, to: 3, slot: 0, from_field: 0 },
            Edge { from: 3, to: 4, slot: 0, from_field: 0 },
            Edge { from: 4, to: 5, slot: 0, from_field: 0 },
            Edge { from: 3, to: 6, slot: 0, from_field: 0 },
        ],
    })
    .expect("valid signal-quality DAG");

    h.run(vec![source]);

    let eq_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_eq.try_iter().collect();
    let ex_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_ex.try_iter().collect();
    let equity = f64_field(&eq_rows, 0);
    let exposure = f64_field(&ex_rows, 0);
    RunReport {
        manifest: RunManifest {
            commit: "streaming-seam-test".to_string(),
            params: vec![
                ("sma_fast".to_string(), 2.0),
                ("sma_slow".to_string(), 4.0),
                ("exposure_scale".to_string(), 0.5),
            ],
            window,
            seed: 0,
            broker: "sim-optimal(pip_size=0.0001)".to_string(),
        },
        metrics: summarize(&equity, &exposure),
    }
}

fn skip_if_no_data(server: &Arc<DataServer>) -> bool {
    if !server.has_symbol(SYMBOL) {
        eprintln!("skip: no local data at {DEFAULT_DATA_PATH} (symbol {SYMBOL} absent)");
        return true;
    }
    false
}

#[test]
fn streaming_close_source_backtests_end_to_end_deterministically() {
    let server = Arc::new(DataServer::new(DEFAULT_DATA_PATH));
    if skip_if_no_data(&server) {
        return;
    }
    let open_close = || {
        M1FieldSource::open(&server, SYMBOL, Some(FROM_MS), Some(TO_MS), M1Field::Close)
            .expect("AAPL.US has data in the 2006-08 window")
    };

    // window bounds for the manifest: drain a source once to read first/last ts.
    let mut probe = open_close();
    let first = probe.peek().expect("non-empty window");
    let mut last = first;
    while let Some((t, _)) = Source::next(&mut probe) {
        last = t;
    }
    let window = (first, last);

    let r1 = run_sample(window, Box::new(open_close()));
    assert!(r1.metrics.total_pips.is_finite(), "a backtest ran over the streamed window");

    // same window streamed again ⇒ bit-identical report (C1).
    let r2 = run_sample(window, Box::new(open_close()));
    assert_eq!(r1.to_json(), r2.to_json());
}

#[test]
fn residency_is_bounded_by_one_chunk_independent_of_window_length() {
    let server = Arc::new(DataServer::new(DEFAULT_DATA_PATH));
    if skip_if_no_data(&server) {
        return;
    }
    let mut src = M1FieldSource::open(&server, SYMBOL, Some(FROM_MS), Some(TO_MS), M1Field::Close)
        .expect("AAPL.US has data in the 2006-08 window");

    // Drive the source the way `run` does (peek to pick, next to pop), sampling
    // resident records at every pull.
    let mut peak = 0usize;
    let mut total = 0usize;
    while src.peek().is_some() {
        peak = peak.max(src.resident_records());
        Source::next(&mut src);
        total += 1;
    }

    // multi-chunk window: more records than a single chunk holds, so the bound
    // below is non-vacuous (a O(window) source would have peaked at `total`).
    assert!(total > CHUNK_SIZE, "window must span multiple chunks (got {total} records)");
    // O(one chunk), NOT O(window): the per-pull ceiling never exceeds one chunk,
    // regardless of how many chunks the window spans.
    assert!(peak <= CHUNK_SIZE, "resident records {peak} exceeded one chunk ({CHUNK_SIZE})");
    assert!(peak > 0, "a non-empty window resided at least one record");
}

#[test]
fn two_field_sources_share_one_window() {
    let server = Arc::new(DataServer::new(DEFAULT_DATA_PATH));
    if skip_if_no_data(&server) {
        return;
    }
    // close + volume over the same window: each is an independent Source pulling
    // its field; both stream to exhaustion (the N-sources-share-the-ts-axis shape).
    let mut close = M1FieldSource::open(&server, SYMBOL, Some(FROM_MS), Some(TO_MS), M1Field::Close)
        .expect("close source");
    let mut volume = M1FieldSource::open(&server, SYMBOL, Some(FROM_MS), Some(TO_MS), M1Field::Volume)
        .expect("volume source");

    let mut n_close = 0usize;
    while let Some((_, v)) = Source::next(&mut close) {
        assert!(matches!(v, Scalar::F64(_)), "close is an f64 column");
        n_close += 1;
    }
    let mut n_volume = 0usize;
    while let Some((_, v)) = Source::next(&mut volume) {
        assert!(matches!(v, Scalar::I64(_)), "volume is an i64 column");
        n_volume += 1;
    }
    assert_eq!(n_close, n_volume, "both fields share the same ts axis ⇒ same count");
    assert!(n_close > 0);
}

(The VecSource import is retained for parity even though this file uses only M1FieldSource; drop it if clippy flags it unused — see Step 3.)

• Step 2: Run the gated streaming tests

Run: cargo test -p aura-ingest --test streaming_seam Expected (where local data is present): PASS, 3 tests (streaming_close_source_backtests_end_to_end_deterministically, residency_is_bounded_by_one_chunk_independent_of_window_length, two_field_sources_share_one_window). Expected (where local data is absent): all 3 print skip: and pass (hermetic).

• Step 3: Lint + unused-import scrub

Run: cargo clippy -p aura-ingest --all-targets -- -D warnings Expected: clean. If VecSource (or any import) is flagged unused in streaming_seam.rs, remove it from the use aura_engine::{...} line.

• Step 4: Full workspace gate

Run: cargo test --workspace Expected: all green.

Run: cargo clippy --workspace --all-targets -- -D warnings Expected: clean.

Run: cargo doc --workspace --no-deps 2>&1 Expected: clean (the new pub items — Source, VecSource, M1Field, M1FieldSource — carry doc comments; no missing-doc or broken-intra-doc-link).