feat(aura-core, aura-engine): Node::initialize — the start-of-stream mirror of finalize

Adds the second lifecycle hook the #283 design revision pairs with
finalize: Node::initialize (default no-op) runs once per node, in
topological order, before the first source value — inside the
deterministic sequence, no within-sim concurrency (C1). Consumers
acquire run resources there (the record consumer opens its streaming
writer); the hook is infallible by signature, so an acquiring node
stores its error, degrades to inert, and surfaces the failure once,
terminally, at finalize.

Every existing node keeps the default. Pinned by the mirror test
run_initializes_every_node_once_before_the_stream (init0,init1
strictly before any eval, exactly once), the shape of the existing
finalize test. RED was observed by temporarily reverting the flush
(test fails with evals-before-inits), then re-applied.

Ledger note for cycle close: C8's lifecycle section gains initialize
as finalize's mirror.

refs #283

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

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