feat(aura-ingest): streaming M1FieldSource — lazy data-server window through the seam

Completes the Source ingestion seam (plan Tasks 3-4): a real, lazily-streamed
data-server source proving the producer seam against the actual data source,
and closing the cycle-0011 deliberate eager-materialization gap.

  - M1Field + a free decode(field, bar) — a pure projection of one M1 bar into
    (epoch-ns ts, Scalar), reusing unix_ms_to_epoch_ns so ms->ns stays the one
    C3 normalization point. Pure ⇒ hermetically unit-tested without an iterator.
  - M1FieldSource: a streaming Source over a data-server M1 window. Holds one
    Arc<[M1Parsed]> chunk (a zero-copy clone of the cache's chunk) + a cursor +
    a pre-decoded head (so peek(&self) is cheap over a &mut next_chunk refill);
    constructs each Scalar per-pull, refills when the chunk drains. Resident
    footprint is O(one chunk), independent of window length.
  - aura-engine moved [dev-dependencies] -> [dependencies] (M1FieldSource is a
    library item implementing aura_engine::Source, not a test-only type).
  - A gated streaming integration test (skips where local data is absent):
      * end-to-end backtest through the seam to a RunReport, C1-deterministic
        (two runs bit-identical);
      * residency predicate — driving the source the way run() does (peek/next)
        over the *unbounded* archive (~1.77M bars locally), peak resident_records
        stays <= one data-server chunk (CHUNK_SIZE) while total >> CHUNK_SIZE:
        O(one chunk), NOT O(window). A O(window) source would peak at `total`;
      * two-field sharing (close + volume over one window, same ts axis).

The eager load_m1_window / close_stream path is kept (still valid for bounded
loads); the gap is closed by a streaming path now existing, not by deleting the
eager one. The residency test drives the full archive rather than the plan's
narrow 2006-08 window, which holds only ~21 bars locally — too few for the
> CHUNK_SIZE non-vacuity precondition; the unbounded archive is the stronger
"independent of window length" proof.

Verified: cargo build/test/clippy --workspace clean; the 3 gated tests ran
against real local data and passed; aura-ingest/aura-engine docs clean (two
pre-existing aura-registry doc-link warnings are from an earlier commit, not
this cycle).

closes #71

crates/aura-ingest/Cargo.toml

@@ -11,8 +11,10 @@ aura-core = { path = "../aura-core" }
 # ingestion edge (transitively chrono + regex + zip). Dependencies are admitted
 # by the amended C16 per-case policy (INDEX.md), not a blanket zero-dep wall.
 data-server = { git = "http://192.168.178.103:3000/Brummel/data-server.git", branch = "main" }
+# M1FieldSource impls aura_engine::Source in the lib, so the engine is a regular
+# (non-dev) dependency.
+aura-engine = { path = "../aura-engine" }
 
 [dev-dependencies]
-# the integration test bootstraps a sample harness + folds a RunReport
-aura-engine = { path = "../aura-engine" }
+# the integration tests bootstrap a sample harness + fold a RunReport
 aura-std = { path = "../aura-std" }

crates/aura-ingest/src/lib.rs

@@ -14,7 +14,7 @@
 
 use aura_core::{Scalar, Timestamp};
 use data_server::records::M1Parsed;
-use data_server::DataServer;
+use data_server::{DataServer, SymbolChunkIter};
 use std::sync::Arc;
 
 /// Normalize data-server's Unix-millisecond time to aura's canonical epoch-ns
@@ -122,6 +122,106 @@ pub fn load_m1_window(
     Some(transpose_m1(&bars))
 }
 
+/// 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)
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -205,4 +305,17 @@ mod tests {
     fn close_stream_empty_on_empty_columns() {
         assert!(transpose_m1(&[]).close_stream().is_empty());
     }
+
+    #[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)));
+    }
 }

crates/aura-ingest/tests/streaming_seam.rs

@@ -0,0 +1,191 @@
+//! 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,
+};
+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";
+// A bounded 2006-08 window in inclusive Unix-ms — enough bars to run the e2e
+// backtest and the two-field sharing test end-to-end (it need not be
+// multi-chunk; the residency test drives the unbounded archive instead).
+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;
+    }
+    // The whole archive (unbounded both ends), not the narrow e2e window: the
+    // non-vacuity assertion below needs > CHUNK_SIZE bars, and early AAPL.US M1
+    // coverage is sparse (a single month can hold far fewer). Driving the full
+    // span is also the strongest form of the "residency independent of window
+    // length" property — the longest window the local data offers.
+    let mut src = M1FieldSource::open(&server, SYMBOL, None, None, M1Field::Close)
+        .expect("AAPL.US has archived data");
+
+    // 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);
+}