diff --git a/crates/aura-core/src/node.rs b/crates/aura-core/src/node.rs
index 520712e..b04f0a8 100644
--- a/crates/aura-core/src/node.rs
+++ b/crates/aura-core/src/node.rs
@@ -379,6 +379,15 @@ pub trait Node {
         "node".to_string()
     }
 
+    /// Start-of-stream hook: `Harness::run` calls it once per node, in
+    /// topological order, before the first source value — the mirror of
+    /// [`Node::finalize`]. A consumer overrides it to acquire run resources
+    /// (e.g. the file a recording sink streams into). Infallible by
+    /// signature: an implementation that can fail stores its error, degrades
+    /// to inert, and surfaces the failure once, terminally, at `finalize`.
+    /// The default is a no-op, so existing nodes are unaffected.
+    fn initialize(&mut self) {}
+
     /// End-of-stream hook: `Harness::run` calls it once per node, in topological
     /// order, after the source loop drains. A folding sink overrides it to flush
     /// its accumulated summary; the default is a no-op, so existing nodes are
diff --git a/crates/aura-engine/src/harness.rs b/crates/aura-engine/src/harness.rs
index 5785ca4..74cdc88 100644
--- a/crates/aura-engine/src/harness.rs
+++ b/crates/aura-engine/src/harness.rs
@@ -519,6 +519,14 @@ impl Harness {
         let mut cycle_id: u64 = 0;
         let mut scratch: Vec<Cell> = Vec::new();
 
+        // start-of-stream: initialize every node once, in topological order,
+        // before the first source value (the mirror of the end-of-stream
+        // finalize flush below). Runs inside the deterministic sequence — no
+        // within-sim concurrency (C1).
+        for &nidx in topo.iter() {
+            nodes[nidx].node.initialize();
+        }
+
         loop {
             // pick the live source head with the smallest (timestamp, source index).
             // strictly-`<` replace + source-order scan preserves C4 tie-breaking.
@@ -2809,6 +2817,52 @@ mod tests {
         assert_eq!(fired, vec![0, 1]);
     }
 
+    struct InitProbe {
+        id: usize,
+        tx: mpsc::Sender<String>,
+        evaled: bool,
+    }
+    impl Node for InitProbe {
+        fn lookbacks(&self) -> Vec<usize> {
+            vec![1]
+        }
+        fn initialize(&mut self) {
+            let _ = self.tx.send(format!("init{}", self.id));
+        }
+        fn eval(&mut self, _ctx: Ctx<'_>) -> Option<&[Cell]> {
+            if !self.evaled {
+                self.evaled = true;
+                let _ = self.tx.send(format!("eval{}", self.id));
+            }
+            None
+        }
+    }
+
+    #[test]
+    fn run_initializes_every_node_once_before_the_stream() {
+        let (tx, rx) = mpsc::channel();
+        // two sink probes, both fed by the single source; no inter-node edges.
+        let mut h = boot(
+            vec![
+                Box::new(InitProbe { id: 0, tx: tx.clone(), evaled: false }),
+                Box::new(InitProbe { id: 1, tx, evaled: false }),
+            ],
+            vec![
+                recorder_sig(&[ScalarKind::F64], Firing::Any),
+                recorder_sig(&[ScalarKind::F64], Firing::Any),
+            ],
+            vec![SourceSpec::raw(ScalarKind::F64, vec![Target { node: 0, slot: 0 }, Target { node: 1, slot: 0 }])],
+            vec![],
+        )
+        .expect("valid");
+        // nothing initialized before the run.
+        assert!(rx.try_recv().is_err());
+        h.run(vec![Box::new(VecSource::new(f64_stream(&[(1, 1.0), (2, 2.0)])))]);
+        // both inits fire, in topo order, strictly before any eval.
+        let fired: Vec<String> = rx.try_iter().collect();
+        assert_eq!(fired, vec!["init0", "init1", "eval0", "eval1"]);
+    }
+
     #[test]
     fn bind_tap_attaches_a_caller_built_sink_and_records_the_tapped_series() {
         // A tap on a producer; bind_tap attaches a caller-built Recorder; the run