diff --git a/crates/aura-cli/src/main.rs b/crates/aura-cli/src/main.rs
index 7c5971b..6ecd0d1 100644
--- a/crates/aura-cli/src/main.rs
+++ b/crates/aura-cli/src/main.rs
@@ -11,7 +11,7 @@ mod render;
 use aura_core::{Firing, Scalar, ScalarKind, Timestamp};
 use aura_engine::{
     f64_field, summarize, Composite, Edge, FlatGraph, GraphBuilder, Harness,
-    RunManifest, RunReport, SourceSpec, SweepFamily, Target,
+    RunManifest, RunReport, SourceSpec, SweepFamily, Target, VecSource,
 };
 use aura_registry::{rank_by, Registry};
 use aura_std::{Ema, Exposure, LinComb, Recorder, SimBroker, Sma, Sub};
@@ -137,7 +137,7 @@ fn run_sample() -> RunReport {
         prices.first().expect("non-empty stream").0,
         prices.last().expect("non-empty stream").0,
     );
-    h.run(vec![prices]);
+    h.run(vec![Box::new(VecSource::new(prices))]);
 
     let eq_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_eq.try_iter().collect();
     let ex_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_ex.try_iter().collect();
@@ -280,7 +280,7 @@ fn sweep_family() -> SweepFamily {
                 prices.first().expect("non-empty stream").0,
                 prices.last().expect("non-empty stream").0,
             );
-            h.run(vec![prices]);
+            h.run(vec![Box::new(VecSource::new(prices))]);
             let equity = f64_field(&rx_eq.try_iter().collect::<Vec<_>>(), 0);
             let exposure = f64_field(&rx_ex.try_iter().collect::<Vec<_>>(), 0);
             let params = space
@@ -459,7 +459,7 @@ fn run_macd() -> RunReport {
         prices.first().expect("non-empty stream").0,
         prices.last().expect("non-empty stream").0,
     );
-    h.run(vec![prices]);
+    h.run(vec![Box::new(VecSource::new(prices))]);
 
     let eq_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_eq.try_iter().collect();
     let ex_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_ex.try_iter().collect();
@@ -524,7 +524,7 @@ mod tests {
             .with("exposure.scale", 0.5)
             .bootstrap()
             .expect("sample blueprint compiles under a valid point");
-        h.run(vec![showcase_prices()]);
+        h.run(vec![Box::new(VecSource::new(showcase_prices()))]);
         assert!(!rx_eq.try_iter().collect::<Vec<_>>().is_empty(), "equity sink drained empty");
         assert!(!rx_ex.try_iter().collect::<Vec<_>>().is_empty(), "exposure sink drained empty");
     }
diff --git a/crates/aura-engine/src/blueprint.rs b/crates/aura-engine/src/blueprint.rs
index 6ef4906..9893e64 100644
--- a/crates/aura-engine/src/blueprint.rs
+++ b/crates/aura-engine/src/blueprint.rs
@@ -812,6 +812,7 @@ fn slot_kind(t: Target, flat_signatures: &[NodeSchema]) -> Result<ScalarKind, Co
 mod tests {
     use super::*;
     use crate::test_fixtures::{composite_sma_cross_harness, synthetic_prices};
+    use crate::VecSource;
     use aura_core::{Ctx, FieldSpec, Firing, NodeSchema, Timestamp};
     use aura_std::{Ema, Exposure, Recorder, SimBroker, Sma, Sub};
     use std::sync::mpsc;
@@ -1020,7 +1021,7 @@ mod tests {
             .with("exposure.scale", 0.5)
             .bootstrap()
             .expect("named binding resolves and bootstraps");
-        named.run(vec![synthetic_prices()]);
+        named.run(vec![Box::new(VecSource::new(synthetic_prices()))]);
         let named_eq = comp_eq.try_iter().collect::<Vec<_>>();
         let named_ex = comp_ex.try_iter().collect::<Vec<_>>();
 
@@ -1028,7 +1029,7 @@ mod tests {
         let mut positional = bp2
             .bootstrap_with_params(vec![Scalar::I64(2), Scalar::I64(4), Scalar::F64(0.5)])
             .expect("positional bootstrap");
-        positional.run(vec![synthetic_prices()]);
+        positional.run(vec![Box::new(VecSource::new(synthetic_prices()))]);
         let pos_eq_v = pos_eq.try_iter().collect::<Vec<_>>();
         let pos_ex_v = pos_ex.try_iter().collect::<Vec<_>>();
 
@@ -1739,14 +1740,14 @@ mod tests {
 
         // (a) today's flat, hand-wired graph
         let (mut flat, flat_eq, flat_ex) = hand_wired_sma_cross_harness();
-        flat.run(vec![prices.clone()]);
+        flat.run(vec![Box::new(VecSource::new(prices.clone()))]);
 
         // (b) the same graph authored as a composite blueprint, compiled
         let (bp, comp_eq, comp_ex) = composite_sma_cross_harness();
         let mut composed = bp
             .bootstrap_with_params(vec![Scalar::I64(2), Scalar::I64(4), Scalar::F64(0.5)])
             .expect("composite blueprint compiles");
-        composed.run(vec![prices]);
+        composed.run(vec![Box::new(VecSource::new(prices))]);
 
         let flat_eq_v = flat_eq.try_iter().collect::<Vec<_>>();
         let flat_ex_v = flat_ex.try_iter().collect::<Vec<_>>();
@@ -1809,7 +1810,7 @@ mod tests {
         let mut h = bp
             .bootstrap_with_params(vec![Scalar::I64(2), Scalar::I64(4)])
             .expect("multi-output composite bootstraps");
-        h.run(vec![prices]);
+        h.run(vec![Box::new(VecSource::new(prices))]);
 
         let a = rx_a.try_iter().collect::<Vec<_>>();
         let b = rx_b.try_iter().collect::<Vec<_>>();
@@ -1826,13 +1827,13 @@ mod tests {
         let (bp, eq, _ex) = composite_sma_cross_harness();
         let mut a = bp.bootstrap_with_params(vec![Scalar::I64(2), Scalar::I64(4), Scalar::F64(0.5)])
             .expect("compiles");
-        a.run(vec![prices.clone()]);
+        a.run(vec![Box::new(VecSource::new(prices.clone()))]);
         let a_eq = eq.try_iter().collect::<Vec<_>>();
 
         let (bp2, eq2, _ex2) = composite_sma_cross_harness();
         let mut b = bp2.bootstrap_with_params(vec![Scalar::I64(5), Scalar::I64(20), Scalar::F64(1.0)])
             .expect("compiles");
-        b.run(vec![prices]);
+        b.run(vec![Box::new(VecSource::new(prices))]);
         let b_eq = eq2.try_iter().collect::<Vec<_>>();
 
         assert!(!a_eq.is_empty() && !b_eq.is_empty(), "both traces populated");
@@ -1870,11 +1871,11 @@ mod tests {
         let (bp, eq, _ex) = composite_sma_cross_harness();
         let mut a = bp.bootstrap_with_params(vec![Scalar::I64(3), Scalar::I64(9), Scalar::F64(0.7)])
             .expect("compiles");
-        a.run(vec![prices.clone()]);
+        a.run(vec![Box::new(VecSource::new(prices.clone()))]);
         let (bp2, eq2, _ex2) = composite_sma_cross_harness();
         let mut b = bp2.bootstrap_with_params(vec![Scalar::I64(3), Scalar::I64(9), Scalar::F64(0.7)])
             .expect("compiles");
-        b.run(vec![prices]);
+        b.run(vec![Box::new(VecSource::new(prices))]);
         assert_eq!(eq.try_iter().collect::<Vec<_>>(), eq2.try_iter().collect::<Vec<_>>());
     }
 
@@ -2392,7 +2393,7 @@ mod tests {
             .with("exposure.scale", 0.5)
             .bootstrap()
             .expect("named (injective) cross resolves by name and bootstraps");
-        h.run(vec![prices]);
+        h.run(vec![Box::new(VecSource::new(prices))]);
         let ex = rx_ex.try_iter().collect::<Vec<_>>();
         assert!(!ex.is_empty(), "the cured, by-name-bound cross must run to a populated exposure trace");
     }
diff --git a/crates/aura-engine/src/harness.rs b/crates/aura-engine/src/harness.rs
index 981253d..fa08953 100644
--- a/crates/aura-engine/src/harness.rs
+++ b/crates/aura-engine/src/harness.rs
@@ -51,6 +51,45 @@ pub struct SourceSpec {
     pub targets: Vec<Target>,
 }
 
+/// 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)
+    }
+}
+
 /// The flat, type-erased, index-wired output of `Composite::compile` — the target
 /// `Harness::bootstrap` consumes (C23's "flat graph", now a named type). `signatures[i]`
 /// is the static signature of `nodes[i]` (gathered from each primitive's builder at
@@ -217,12 +256,12 @@ impl Harness {
     /// its record to a destination it holds (out of graph) inside `eval`; the
     /// engine only routes in-graph edges and is oblivious to the side effect.
     /// Allocates nothing per cycle beyond the reused scratch buffer.
-    pub fn run(&mut self, streams: Vec<Vec<(Timestamp, Scalar)>>) {
+    pub fn run(&mut self, mut sources_in: Vec<Box<dyn Source>>) {
         assert_eq!(
-            streams.len(),
+            sources_in.len(),
             self.sources.len(),
-            "run: one stream per source required (got {} streams for {} sources)",
-            streams.len(),
+            "run: one source per SourceSpec required (got {} sources for {} specs)",
+            sources_in.len(),
             self.sources.len()
         );
 
@@ -230,31 +269,31 @@ impl Harness {
         // 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() {
+            // 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),
-                        Some(p) => {
-                            if stream[cursor[s]].0 < streams[p][cursor[p]].0 {
-                                pick = Some(s);
-                            }
-                        }
+                        None => pick = Some((s, ts)),
+                        Some((_, best)) if ts < best => pick = Some((s, ts)),
+                        _ => {}
                     }
                 }
             }
             let s = match pick {
-                Some(s) => s,
-                None => break, // all streams exhausted
+                Some((s, _)) => s,
+                None => break, // all sources exhausted
             };
-            let (ts, value) = streams[s][cursor[s]];
-            cursor[s] += 1;
+            // `&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;
 
             // forward the source value into its target slots, stamping freshness
@@ -356,6 +395,32 @@ mod tests {
         points.iter().map(|&(t, v)| (Timestamp(t), Scalar::F64(v))).collect()
     }
 
+    #[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);
+    }
+
     /// One f64 input port with the given firing policy (lookback 1 is the bootstrap
     /// default for these test fixtures, supplied by their `lookbacks()`).
     fn f64_port(firing: Firing) -> PortSpec {
@@ -591,7 +656,7 @@ mod tests {
             vec![Edge { from: 0, to: 1, slot: 0, from_field: 0 }],
         )
         .expect("valid");
-        h.run(vec![f64_stream(&[(1, 1.0), (2, 2.0), (3, 3.0), (4, 4.0), (5, 5.0)])]);
+        h.run(vec![Box::new(VecSource::new(f64_stream(&[(1, 1.0), (2, 2.0), (3, 3.0), (4, 4.0), (5, 5.0)])))]);
         let got: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
         // SMA(3) warms at cycle 3; the recorder captures only fired cycles, each
         // tagged with the cycle's timestamp (sparse — no None hold-rows).
@@ -639,7 +704,7 @@ mod tests {
 
         let (tx, rx) = mpsc::channel();
         let mut h = build(tx);
-        h.run(vec![prices.clone()]);
+        h.run(vec![Box::new(VecSource::new(prices.clone()))]);
         let out: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
         // Sub fires once SMA(4) is warm (cycle 4): 15-13, 17-15, 19-17 -> 2.
         assert_eq!(
@@ -654,7 +719,7 @@ mod tests {
         // determinism (C1): a second identical run drains a bit-identical stream.
         let (tx2, rx2) = mpsc::channel();
         let mut h2 = build(tx2);
-        h2.run(vec![prices]);
+        h2.run(vec![Box::new(VecSource::new(prices))]);
         let out2: Vec<(Timestamp, Vec<Scalar>)> = rx2.try_iter().collect();
         assert_eq!(out2, out);
     }
@@ -686,7 +751,7 @@ mod tests {
 
         let (tx, rx) = mpsc::channel();
         let mut h = build(tx);
-        h.run(vec![s0.clone(), s1.clone()]);
+        h.run(vec![Box::new(VecSource::new(s0.clone())), Box::new(VecSource::new(s1.clone()))]);
         let out: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
         // holds s1=100 across t=3 and the t=4 s0-cycle; emits on every tick once warm.
         assert_eq!(
@@ -701,7 +766,7 @@ mod tests {
 
         let (tx2, rx2) = mpsc::channel();
         let mut h2 = build(tx2);
-        h2.run(vec![s0, s1]);
+        h2.run(vec![Box::new(VecSource::new(s0)), Box::new(VecSource::new(s1))]);
         let out2: Vec<(Timestamp, Vec<Scalar>)> = rx2.try_iter().collect();
         assert_eq!(out2, out); // deterministic
     }
@@ -732,7 +797,7 @@ mod tests {
 
         let (tx, rx) = mpsc::channel();
         let mut h = build(tx);
-        h.run(vec![s0.clone(), s1.clone()]);
+        h.run(vec![Box::new(VecSource::new(s0.clone())), Box::new(VecSource::new(s1.clone()))]);
         let out: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
         // records ONLY at t=2 and t=4 where both inputs share the timestamp.
         assert_eq!(
@@ -745,7 +810,7 @@ mod tests {
 
         let (tx2, rx2) = mpsc::channel();
         let mut h2 = build(tx2);
-        h2.run(vec![s0, s1]);
+        h2.run(vec![Box::new(VecSource::new(s0)), Box::new(VecSource::new(s1))]);
         let out2: Vec<(Timestamp, Vec<Scalar>)> = rx2.try_iter().collect();
         assert_eq!(out2, out); // deterministic
     }
@@ -786,7 +851,7 @@ mod tests {
 
         let (tx, rx) = mpsc::channel();
         let mut h = build(tx);
-        h.run(vec![prices.clone()]);
+        h.run(vec![Box::new(VecSource::new(prices.clone()))]);
         let out: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
         // SMA(4) warms at cycle 4; from then both paths emit each cycle at the same
         // timestamp, so the barrier fires: SMA(2)+SMA(4) = 15+13, 17+15, 19+17.
@@ -801,7 +866,7 @@ mod tests {
 
         let (tx2, rx2) = mpsc::channel();
         let mut h2 = build(tx2);
-        h2.run(vec![prices]);
+        h2.run(vec![Box::new(VecSource::new(prices))]);
         let out2: Vec<(Timestamp, Vec<Scalar>)> = rx2.try_iter().collect();
         assert_eq!(out2, out);
     }
@@ -831,7 +896,7 @@ mod tests {
         let s1 = f64_stream(&[(2, 200.0)]);           // in1 (barrier)
         let s2 = f64_stream(&[(1, 1.0), (3, 3.0)]);   // in2 (as-of)
 
-        h.run(vec![s0, s1, s2]);
+        h.run(vec![Box::new(VecSource::new(s0)), Box::new(VecSource::new(s1)), Box::new(VecSource::new(s2))]);
         let out: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
         // c3: barrier pair completes at t=2, holds c=1 -> 221. c4: as-of input
         // ticks at t=3, holds the pair -> 223. c1 filters; c2,c5 hold (no record).
@@ -897,13 +962,13 @@ mod tests {
     }
 
     /// Build five timestamp-aligned f64 sources feeding Ohlcv's five barrier slots.
-    fn ohlcv_streams() -> Vec<Vec<(Timestamp, Scalar)>> {
+    fn ohlcv_streams() -> Vec<Box<dyn Source>> {
         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
+            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
         ]
     }
 
@@ -1119,7 +1184,7 @@ mod tests {
             ],
         )
         .expect("valid DAG");
-        h.run(vec![f64_stream(&[(1, 10.0), (2, 12.0), (3, 14.0), (4, 16.0), (5, 18.0)])]);
+        h.run(vec![Box::new(VecSource::new(f64_stream(&[(1, 10.0), (2, 12.0), (3, 14.0), (4, 16.0), (5, 18.0)])))]);
         let fast: Vec<(Timestamp, Vec<Scalar>)> = rx_fast.try_iter().collect();
         let slow: Vec<(Timestamp, Vec<Scalar>)> = rx_slow.try_iter().collect();
         // SMA(2) warms at cycle 2, SMA(4) at cycle 4 — two different-rate streams.
@@ -1170,11 +1235,11 @@ mod tests {
         )
         .expect("valid");
         h.run(vec![
-            f64_stream(&[(1, 10.0)]),
-            f64_stream(&[(1, 15.0)]),
-            f64_stream(&[(1, 8.0)]),
-            f64_stream(&[(1, 12.0)]),
-            f64_stream(&[(1, 100.0)]),
+            Box::new(VecSource::new(f64_stream(&[(1, 10.0)]))),
+            Box::new(VecSource::new(f64_stream(&[(1, 15.0)]))),
+            Box::new(VecSource::new(f64_stream(&[(1, 8.0)]))),
+            Box::new(VecSource::new(f64_stream(&[(1, 12.0)]))),
+            Box::new(VecSource::new(f64_stream(&[(1, 100.0)]))),
         ]);
         let out: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
         assert_eq!(out.len(), 1);
@@ -1217,10 +1282,10 @@ mod tests {
         )
         .expect("valid");
         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)))],
+            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)))])),
         ]);
         let out: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
         assert_eq!(
@@ -1254,7 +1319,7 @@ mod tests {
             vec![Edge { from: 0, to: 1, slot: 0, from_field: 0 }],
         )
         .expect("valid");
-        h.run(vec![f64_stream(&[(1, 10.0), (2, 20.0), (3, 30.0)])]);
+        h.run(vec![Box::new(VecSource::new(f64_stream(&[(1, 10.0), (2, 20.0), (3, 30.0)])))]);
         let tapped: Vec<(Timestamp, Vec<Scalar>)> = rx_tap.try_iter().collect();
         let downstream: Vec<(Timestamp, Vec<Scalar>)> = rx_down.try_iter().collect();
         let expected = vec![
@@ -1289,12 +1354,12 @@ mod tests {
 
         let (tx_a, rx_a) = mpsc::channel();
         let mut a = build(tx_a);
-        a.run(vec![prices.clone()]);
+        a.run(vec![Box::new(VecSource::new(prices.clone()))]);
         let run_a: Vec<(Timestamp, Vec<Scalar>)> = rx_a.try_iter().collect();
 
         let (tx_b, rx_b) = mpsc::channel();
         let mut b = build(tx_b);
-        b.run(vec![prices]);
+        b.run(vec![Box::new(VecSource::new(prices))]);
         let run_b: Vec<(Timestamp, Vec<Scalar>)> = rx_b.try_iter().collect();
 
         assert_eq!(run_a, run_b);
@@ -1324,7 +1389,7 @@ mod tests {
         .expect("valid");
         let s0 = f64_stream(&[(1, 10.0), (2, 20.0), (3, 30.0), (4, 40.0)]);
         let s1 = f64_stream(&[(2, 100.0), (4, 200.0)]);
-        h.run(vec![s0, s1]);
+        h.run(vec![Box::new(VecSource::new(s0)), Box::new(VecSource::new(s1))]);
         let out: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
         // records ONLY at t=2 and t=4 (both inputs coincide); holds otherwise.
         assert_eq!(
@@ -1359,7 +1424,7 @@ mod tests {
         .expect("valid");
         let s0 = f64_stream(&[(1, 10.0), (2, 20.0), (3, 30.0), (4, 40.0)]);
         let s1 = f64_stream(&[(2, 100.0), (4, 200.0)]);
-        h.run(vec![s0, s1]);
+        h.run(vec![Box::new(VecSource::new(s0)), Box::new(VecSource::new(s1))]);
         let out: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
         // from t=2 on, records every cycle holding the stale input; two cycles fall
         // on t=4 (the s0 tick then the s1 tick).
@@ -1444,7 +1509,7 @@ mod tests {
         let (b1, rb) = mpsc::channel();
         let (c1, rc) = mpsc::channel();
         let mut h = build(a1, b1, c1);
-        h.run(vec![prices.clone()]);
+        h.run(vec![Box::new(VecSource::new(prices.clone()))]);
         let s1: Vec<(Timestamp, Vec<Scalar>)> = ra.try_iter().collect();
         let s2: Vec<(Timestamp, Vec<Scalar>)> = rb.try_iter().collect();
         let s3: Vec<(Timestamp, Vec<Scalar>)> = rc.try_iter().collect();
@@ -1463,7 +1528,7 @@ mod tests {
         let (b2, rb2) = mpsc::channel();
         let (c2, rc2) = mpsc::channel();
         let mut h2 = build(a2, b2, c2);
-        h2.run(vec![prices]);
+        h2.run(vec![Box::new(VecSource::new(prices))]);
         assert_eq!(ra2.try_iter().collect::<Vec<_>>(), expected);
         assert_eq!(rb2.try_iter().collect::<Vec<_>>(), expected);
         assert_eq!(rc2.try_iter().collect::<Vec<_>>(), expected);
@@ -1509,7 +1574,7 @@ mod tests {
         let (tr, rr) = mpsc::channel();
         let (ts, rs) = mpsc::channel();
         let mut h = build(tr, ts);
-        h.run(vec![prices.clone()]);
+        h.run(vec![Box::new(VecSource::new(prices.clone()))]);
         let raw: Vec<(Timestamp, Vec<Scalar>)> = rr.try_iter().collect();
         let sub: Vec<(Timestamp, Vec<Scalar>)> = rs.try_iter().collect();
         // SMA(2): 11,13,15,17,19 at cycles 2..6 (raw tap).
@@ -1532,7 +1597,7 @@ mod tests {
         let (tr2, rr2) = mpsc::channel();
         let (ts2, rs2) = mpsc::channel();
         let mut h2 = build(tr2, ts2);
-        h2.run(vec![prices]);
+        h2.run(vec![Box::new(VecSource::new(prices))]);
         assert_eq!(rr2.try_iter().collect::<Vec<_>>(), raw_expected); // deterministic
         assert_eq!(rs2.try_iter().collect::<Vec<_>>(), sub_expected);
     }
@@ -1576,7 +1641,7 @@ mod tests {
         let (ta, rany) = mpsc::channel();
         let (tb, rbar) = mpsc::channel();
         let mut h = build(ta, tb);
-        h.run(vec![a.clone(), b.clone()]);
+        h.run(vec![Box::new(VecSource::new(a.clone())), Box::new(VecSource::new(b.clone()))]);
         let any: Vec<(Timestamp, Vec<Scalar>)> = rany.try_iter().collect();
         let bar: Vec<(Timestamp, Vec<Scalar>)> = rbar.try_iter().collect();
         // As-of tap records every SMA(2) fire: 11@t2, 13@t3, 15@t4.
@@ -1597,7 +1662,7 @@ mod tests {
         let (ta2, rany2) = mpsc::channel();
         let (tb2, rbar2) = mpsc::channel();
         let mut h2 = build(ta2, tb2);
-        h2.run(vec![a, b]);
+        h2.run(vec![Box::new(VecSource::new(a)), Box::new(VecSource::new(b))]);
         assert_eq!(rany2.try_iter().collect::<Vec<_>>(), any_expected); // deterministic
         assert_eq!(rbar2.try_iter().collect::<Vec<_>>(), bar_expected);
     }
@@ -1637,7 +1702,7 @@ mod tests {
 
         let (tx, rx) = mpsc::channel();
         let mut h = build(tx);
-        h.run(vec![prices.clone()]);
+        h.run(vec![Box::new(VecSource::new(prices.clone()))]);
         let out: Vec<(Timestamp, Vec<Scalar>)> = rx.try_iter().collect();
         // stage1 (c2..c5): 3,5,7,9. stage2 (c3..): 4,6,8. stage3 (c4..): 5,7.
         let expected = vec![
@@ -1648,7 +1713,7 @@ mod tests {
 
         let (tx2, rx2) = mpsc::channel();
         let mut h2 = build(tx2);
-        h2.run(vec![prices]);
+        h2.run(vec![Box::new(VecSource::new(prices))]);
         assert_eq!(rx2.try_iter().collect::<Vec<_>>(), expected); // deterministic
     }
 
@@ -1698,7 +1763,7 @@ mod tests {
         let (b, rb) = mpsc::channel();
         let (c, rc) = mpsc::channel();
         let mut h = build(a, b, c);
-        h.run(vec![prices.clone()]);
+        h.run(vec![Box::new(VecSource::new(prices.clone()))]);
         let w2: Vec<(Timestamp, Vec<Scalar>)> = ra.try_iter().collect();
         let w3: Vec<(Timestamp, Vec<Scalar>)> = rb.try_iter().collect();
         let w4: Vec<(Timestamp, Vec<Scalar>)> = rc.try_iter().collect();
@@ -1725,7 +1790,7 @@ mod tests {
         let (b2, rb2) = mpsc::channel();
         let (c2, rc2) = mpsc::channel();
         let mut h2 = build(a2, b2, c2);
-        h2.run(vec![prices]);
+        h2.run(vec![Box::new(VecSource::new(prices))]);
         assert_eq!(ra2.try_iter().collect::<Vec<_>>(), e2); // deterministic
         assert_eq!(rb2.try_iter().collect::<Vec<_>>(), e3);
         assert_eq!(rc2.try_iter().collect::<Vec<_>>(), e4);
@@ -1785,7 +1850,7 @@ mod tests {
         let (ts, rs) = mpsc::channel();
         let (ti, ri) = mpsc::channel();
         let mut h = build(tr, ts, ti);
-        h.run(vec![prices.clone(), counts.clone()]);
+        h.run(vec![Box::new(VecSource::new(prices.clone())), Box::new(VecSource::new(counts.clone()))]);
         let raw: Vec<(Timestamp, Vec<Scalar>)> = rr.try_iter().collect();
         let sub: Vec<(Timestamp, Vec<Scalar>)> = rs.try_iter().collect();
         let i64s: Vec<(Timestamp, Vec<Scalar>)> = ri.try_iter().collect();
@@ -1814,7 +1879,7 @@ mod tests {
         let (ts2, rs2) = mpsc::channel();
         let (ti2, ri2) = mpsc::channel();
         let mut h2 = build(tr2, ts2, ti2);
-        h2.run(vec![prices, counts]);
+        h2.run(vec![Box::new(VecSource::new(prices)), Box::new(VecSource::new(counts))]);
         assert_eq!(rr2.try_iter().collect::<Vec<_>>(), raw_expected); // deterministic
         assert_eq!(rs2.try_iter().collect::<Vec<_>>(), sub_expected);
         assert_eq!(ri2.try_iter().collect::<Vec<_>>(), i64_expected);
@@ -1858,13 +1923,13 @@ mod tests {
         )
         .expect("valid signal-quality harness");
 
-        h.run(vec![f64_stream(&[
+        h.run(vec![Box::new(VecSource::new(f64_stream(&[
             (1, 1.0000),
             (2, 1.0010),
             (3, 1.0025),
             (4, 1.0020),
             (5, 1.0040),
-        ])]);
+        ])))]);
 
         let equity: Vec<(Timestamp, Vec<Scalar>)> = rx_eq.try_iter().collect();
         // broker fires every cycle (price fresh each tick): five records, ts 1..=5
@@ -1923,13 +1988,13 @@ mod tests {
                 ],
             )
             .expect("valid harness");
-            h.run(vec![f64_stream(&[
+            h.run(vec![Box::new(VecSource::new(f64_stream(&[
                 (1, 1.0000),
                 (2, 1.0010),
                 (3, 1.0025),
                 (4, 1.0020),
                 (5, 1.0040),
-            ])]);
+            ])))]);
             rx.try_iter().collect::<Vec<(Timestamp, Vec<Scalar>)>>()
         };
         assert_eq!(build(), build());
diff --git a/crates/aura-engine/src/lib.rs b/crates/aura-engine/src/lib.rs
index 5f97c1d..4b5d7b4 100644
--- a/crates/aura-engine/src/lib.rs
+++ b/crates/aura-engine/src/lib.rs
@@ -22,8 +22,7 @@
 //!   pip-equity + exposure streams into [`RunMetrics`]; [`f64_field`] bridges a
 //!   recording sink's `Vec<Scalar>` rows to it.
 //!
-//! Still to come (subsequent cycles): the `Source` trait + data-server ingestion
-//! and source-native time normalization (C3/C11), the broker-independent
+//! Still to come (subsequent cycles): the broker-independent
 //! position-event output and downstream broker nodes (C10), and the param
 //! injection + orchestration axes of the atomic sim unit
 //! (`(topology + params + data-window + seed)`, swept by optimize / walk-forward
@@ -44,7 +43,7 @@ pub use blueprint::{
 };
 pub use builder::{BuildError, GraphBuilder, InPort, NodeHandle, OutPort, RoleHandle};
 pub use graph_model::model_to_json;
-pub use harness::{BootstrapError, Edge, FlatGraph, Harness, SourceSpec, Target};
+pub use harness::{BootstrapError, Edge, FlatGraph, Harness, Source, SourceSpec, Target, VecSource};
 pub use report::{f64_field, summarize, RunManifest, RunMetrics, RunReport};
 pub use sweep::{sweep, GridSpace, SweepError, SweepFamily, SweepPoint};
 // #29: re-export the core scalar vocabulary a Blueprint builder needs
diff --git a/crates/aura-engine/src/report.rs b/crates/aura-engine/src/report.rs
index 838a8b0..cea3189 100644
--- a/crates/aura-engine/src/report.rs
+++ b/crates/aura-engine/src/report.rs
@@ -145,7 +145,7 @@ pub fn f64_field(rows: &[(Timestamp, Vec<Scalar>)], field: usize) -> Vec<(Timest
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::{Edge, FlatGraph, Harness, SourceSpec, Target};
+    use crate::{Edge, FlatGraph, Harness, SourceSpec, Target, VecSource};
     use aura_core::{Firing, NodeSchema, PortSpec, ScalarKind};
     use aura_std::{Exposure, Recorder, SimBroker, Sma, Sub};
     use std::sync::mpsc;
@@ -220,13 +220,13 @@ mod tests {
 
     fn run_once() -> RunReport {
         let (mut h, rx_eq, rx_ex) = build_two_sink_harness();
-        h.run(vec![f64_stream(&[
+        h.run(vec![Box::new(VecSource::new(f64_stream(&[
             (1, 1.0000),
             (2, 1.0010),
             (3, 1.0025),
             (4, 1.0020),
             (5, 1.0040),
-        ])]);
+        ])))]);
         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);
diff --git a/crates/aura-engine/src/sweep.rs b/crates/aura-engine/src/sweep.rs
index b1dabb2..b384a65 100644
--- a/crates/aura-engine/src/sweep.rs
+++ b/crates/aura-engine/src/sweep.rs
@@ -170,7 +170,7 @@ where
 mod tests {
     use super::*;
     use crate::test_fixtures::{composite_sma_cross_harness, synthetic_prices};
-    use crate::{f64_field, summarize, RunManifest};
+    use crate::{f64_field, summarize, RunManifest, VecSource};
     use aura_core::{ParamSpec, Scalar, ScalarKind, Timestamp};
 
     fn i64_space(n: usize) -> Vec<ParamSpec> {
@@ -260,7 +260,7 @@ mod tests {
         let mut h = bp
             .bootstrap_with_params(point.to_vec())
             .expect("grid points are kind-checked against param_space");
-        h.run(vec![synthetic_prices()]);
+        h.run(vec![Box::new(VecSource::new(synthetic_prices()))]);
         let equity = f64_field(&rx_eq.try_iter().collect::<Vec<_>>(), 0);
         let exposure = f64_field(&rx_ex.try_iter().collect::<Vec<_>>(), 0);
         RunReport {
diff --git a/crates/aura-ingest/tests/real_bars.rs b/crates/aura-ingest/tests/real_bars.rs
index ad1893c..6674509 100644
--- a/crates/aura-ingest/tests/real_bars.rs
+++ b/crates/aura-ingest/tests/real_bars.rs
@@ -10,6 +10,7 @@ use std::sync::Arc;
 use aura_core::{Firing, NodeSchema, PortSpec, Scalar, ScalarKind, Timestamp};
 use aura_engine::{
     f64_field, summarize, Edge, FlatGraph, Harness, RunManifest, RunReport, SourceSpec, Target,
+    VecSource,
 };
 use aura_ingest::load_m1_window;
 use aura_std::{Exposure, Recorder, SimBroker, Sma, Sub};
@@ -69,7 +70,7 @@ fn run_sample_over(prices: Vec<(Timestamp, Scalar)>) -> RunReport {
         prices.first().map(|&(t, _)| t).unwrap_or(Timestamp(0)),
         prices.last().map(|&(t, _)| t).unwrap_or(Timestamp(0)),
     );
-    h.run(vec![prices]);
+    h.run(vec![Box::new(VecSource::new(prices))]);
 
     let eq_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_eq.try_iter().collect();
     let ex_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_ex.try_iter().collect();