From 74324d178f6b4b9f6f0f2baab1454fabc16a33a3 Mon Sep 17 00:00:00 2001 From: Brummel Date: Thu, 18 Jun 2026 14:55:30 +0200 Subject: [PATCH] =?UTF-8?q?feat(aura-engine):=20join=5Fon=5Fts=20=E2=80=94?= =?UTF-8?q?=20timestamp=20join=20for=20multi-tap=20sink=20traces?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) 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 | 2 +- crates/aura-engine/src/report.rs | 108 +++++++++++++++++++++++++++++++ 2 files changed, 109 insertions(+), 1 deletion(-) diff --git a/crates/aura-engine/src/lib.rs b/crates/aura-engine/src/lib.rs index 86f57f2..863a853 100644 --- a/crates/aura-engine/src/lib.rs +++ b/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, }; diff --git a/crates/aura-engine/src/report.rs b/crates/aura-engine/src/report.rs index 6d75b42..afd73c3 100644 --- a/crates/aura-engine/src/report.rs +++ b/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)], 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, + pub sides: Vec>>, +} + +/// 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)], + sides: &[&[(Timestamp, Vec)]], +) -> Vec { + let side_maps: Vec>> = 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![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)])); + } }