feat(aura-engine): join_on_ts — timestamp join for multi-tap sink traces

Ship a generic, spine-anchored join_on_ts helper + JoinedRow in the report
layer (sibling of summarize/f64_field), re-exported from lib.rs. It fuses N
recorded (Timestamp, Vec<Scalar>) tap streams on the recorded timestamp:
exactly one JoinedRow per spine entry, each side looked up by ts as
Some(row)/None where it did not fire. Option-per-side semantics keep the
engine honest — it reports presence; the consumer interprets absence (the
0.0/-1/false defaults are consumer truths, not engine ones). C8/C18 (post-run
reduction over recorded sink output), C3 (no in-graph join), C1 (one row per
ts, documented precondition).

Two unit tests pin the contract: cardinality-misalignment alignment (the #93
shape — a spine bar a side tap skipped resolves to None, not a zip-misalign)
and the spine-anchored drop of a side row at a non-spine ts. RED accepted on
the grounding-check record (both symbols were absent pre-cycle, so the tests
are RED-by-construction) and verified green here.

drain_trace's migration onto this helper + the Recorder doc note follow.

refs #93

crates/aura-engine/src/lib.rs

@@ -61,7 +61,7 @@ pub use harness::{
     window_of, BootstrapError, Edge, FlatGraph, Harness, Source, SourceSpec, SyntheticSpec, Target,
     VecSource,
 };
-pub use report::{f64_field, summarize, RunManifest, RunMetrics, RunReport};
+pub use report::{f64_field, join_on_ts, summarize, JoinedRow, RunManifest, RunMetrics, RunReport};
 pub use sweep::{
     sweep, GridSpace, ParamRange, RandomSpace, Space, SweepError, SweepFamily, SweepPoint,
 };

crates/aura-engine/src/report.rs

@@ -9,6 +9,7 @@
 //! and on-disk shapes coincide.
 
 use aura_core::{Scalar, ScalarKind, Timestamp};
+use std::collections::HashMap;
 
 /// Summary metrics reduced from a run's recorded streams — the `-> metrics`
 /// half of C12's atomic sim unit. Pure function of the recorded streams.
@@ -143,6 +144,48 @@ pub fn f64_field(rows: &[(Timestamp, Vec<Scalar>)], field: usize) -> Vec<(Timest
         .collect()
 }
 
+/// One spine row joined with each side stream's row recorded at the same
+/// timestamp. `sides` is parallel to the `sides` argument of [`join_on_ts`]; an
+/// entry is `None` where that side did not fire at this spine timestamp.
+#[derive(Clone, Debug, PartialEq)]
+pub struct JoinedRow {
+    pub ts: Timestamp,
+    pub spine: Vec<Scalar>,
+    pub sides: Vec<Option<Vec<Scalar>>>,
+}
+
+/// Join recording-sink tap streams on their recorded timestamp (C8/C18: a post-run
+/// reduction over recorded sink output; C3: NOT an in-graph join).
+///
+/// `spine` defines the row set — exactly one [`JoinedRow`] per spine entry, in
+/// spine order. Each side stream is looked up by timestamp: `Some(row)` where it
+/// fired at that timestamp, `None` where it did not. The helper does not interpret
+/// a row's columns (it returns each whole); the caller maps `None` to whatever
+/// default its column means.
+///
+/// Precondition (C1): each stream has at most one row per timestamp — a sink fires
+/// at most once per cycle and cycles have unique timestamps. A duplicate timestamp
+/// within one stream resolves last-write-wins. A side row whose timestamp is absent
+/// from the spine is dropped (the spine defines the rows).
+pub fn join_on_ts(
+    spine: &[(Timestamp, Vec<Scalar>)],
+    sides: &[&[(Timestamp, Vec<Scalar>)]],
+) -> Vec<JoinedRow> {
+    let side_maps: Vec<HashMap<i64, &Vec<Scalar>>> = sides
+        .iter()
+        .map(|s| s.iter().map(|(t, row)| (t.0, row)).collect())
+        .collect();
+
+    spine
+        .iter()
+        .map(|(ts, row)| JoinedRow {
+            ts: *ts,
+            spine: row.clone(),
+            sides: side_maps.iter().map(|m| m.get(&ts.0).map(|r| (*r).clone())).collect(),
+        })
+        .collect()
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -402,4 +445,69 @@ mod tests {
         let back: RunReport = serde_json::from_str(&json).expect("deserialize RunReport");
         assert_eq!(back, report);
     }
+
+    #[test]
+    fn join_on_ts_aligns_streams_of_different_cardinality() {
+        // spine fires every bar; side A is one row shorter (no ts 10, like cold
+        // Delay(1) on the first bar); side B fires on a subset (only ts 20, 40,
+        // like a Session filter before the open).
+        let spine = vec![
+            (Timestamp(10), vec![Scalar::f64(1.0)]),
+            (Timestamp(20), vec![Scalar::f64(2.0)]),
+            (Timestamp(30), vec![Scalar::f64(3.0)]),
+            (Timestamp(40), vec![Scalar::f64(4.0)]),
+        ];
+        let side_a = vec![
+            (Timestamp(20), vec![Scalar::bool(true)]),
+            (Timestamp(30), vec![Scalar::bool(false)]),
+            (Timestamp(40), vec![Scalar::bool(true)]),
+        ];
+        let side_b = vec![
+            (Timestamp(20), vec![Scalar::i64(0)]),
+            (Timestamp(40), vec![Scalar::i64(2)]),
+        ];
+
+        let joined = join_on_ts(&spine, &[&side_a, &side_b]);
+
+        // one row per spine entry, in spine order
+        assert_eq!(joined.len(), 4);
+        assert_eq!(
+            joined.iter().map(|j| j.ts).collect::<Vec<_>>(),
+            vec![Timestamp(10), Timestamp(20), Timestamp(30), Timestamp(40)]
+        );
+
+        // ts 10: spine present, both sides absent (the zip-by-index misalignment case)
+        assert_eq!(joined[0].spine, vec![Scalar::f64(1.0)]);
+        assert_eq!(joined[0].sides[0], None);
+        assert_eq!(joined[0].sides[1], None);
+
+        // ts 20: both sides present and aligned to THIS ts
+        assert_eq!(joined[1].sides[0], Some(vec![Scalar::bool(true)]));
+        assert_eq!(joined[1].sides[1], Some(vec![Scalar::i64(0)]));
+
+        // ts 30: side A present, side B absent
+        assert_eq!(joined[2].sides[0], Some(vec![Scalar::bool(false)]));
+        assert_eq!(joined[2].sides[1], None);
+
+        // ts 40: both present
+        assert_eq!(joined[3].sides[0], Some(vec![Scalar::bool(true)]));
+        assert_eq!(joined[3].sides[1], Some(vec![Scalar::i64(2)]));
+    }
+
+    #[test]
+    fn join_on_ts_drops_side_rows_absent_from_spine() {
+        // a side row whose ts is not in the spine is dropped — the spine defines
+        // the row set.
+        let spine = vec![(Timestamp(10), vec![Scalar::f64(1.0)])];
+        let side = vec![
+            (Timestamp(10), vec![Scalar::i64(7)]),
+            (Timestamp(99), vec![Scalar::i64(8)]), // ts 99 absent from spine -> dropped
+        ];
+
+        let joined = join_on_ts(&spine, &[&side]);
+
+        assert_eq!(joined.len(), 1);
+        assert_eq!(joined[0].ts, Timestamp(10));
+        assert_eq!(joined[0].sides[0], Some(vec![Scalar::i64(7)]));
+    }
 }