diff --git a/docs/plans/0053-multi-tap-ts-join.md b/docs/plans/0053-multi-tap-ts-join.md deleted file mode 100644 index a9258ed..0000000 --- a/docs/plans/0053-multi-tap-ts-join.md +++ /dev/null @@ -1,385 +0,0 @@ -# Multi-tap trace join on timestamp — Implementation Plan - -> **Parent spec:** `docs/specs/0053-multi-tap-ts-join.md` -> -> **For agentic workers:** REQUIRED SUB-SKILL: use the `implement` skill to run -> this plan. Steps use `- [ ]` checkboxes for tracking. - -**Goal:** Ship a generic, spine-anchored `join_on_ts` helper (Option-per-side) in -`aura-engine`'s report layer, refactor `drain_trace` onto it, and document the -multi-tap cadence rule on the `Recorder` API — resolving the #93 zip-by-index panic. - -**Architecture:** `join_on_ts` + `JoinedRow` are a post-run pure reduction over -recorded sink streams, siblings of `summarize`/`f64_field` in -`crates/aura-engine/src/report.rs`, re-exported from `lib.rs`. The consumer-side -`drain_trace` in the GER40 example becomes a thin `BarTrace` mapping over the -helper; its caller-side defaults (`0.0`/`-1`/`false`) are unchanged. The Recorder -note is plain prose (no cross-crate rustdoc link — `aura-std` does not depend on -`aura-engine`). - -**Tech Stack:** Rust; `aura-engine` (report.rs, lib.rs), `aura-ingest` -(examples/shared/breakout_real.rs), `aura-std` (recorder.rs); `aura-core` -`Scalar`/`Timestamp`. - -**Files this plan creates or modifies:** - -- Modify: `crates/aura-engine/src/report.rs` — add `use std::collections::HashMap;` - at the top; insert `JoinedRow` + `join_on_ts` between `f64_field` (ends :144) and - `#[cfg(test)] mod tests` (:146); append two `#[test]` fns at the end of the test - module (before its closing brace at :405). -- Modify: `crates/aura-engine/src/lib.rs:64` — extend the `report` re-export. -- Modify: `crates/aura-ingest/examples/shared/breakout_real.rs` — add `join_on_ts` - to the `aura_engine` use block (:42-45); replace the `drain_trace` body (:354-381). -- Modify: `crates/aura-std/src/recorder.rs:13-16` — append a prose cadence/join note - to the `Recorder` struct doc. -- Test: `crates/aura-engine/src/report.rs` — `join_on_ts_aligns_streams_of_different_cardinality`, - `join_on_ts_drops_side_rows_absent_from_spine` (the two RED unit tests). -- Gate (unchanged, behaviour-preservation): `crates/aura-ingest/tests/ger40_breakout_real.rs`, - `crates/aura-ingest/tests/ger40_breakout_blueprint.rs`, - `crates/aura-ingest/tests/open_ohlc_seam.rs` (a fifth `drain_trace` consumer at - :29 — recon-flagged; it lands under the same `cargo test` gate, no edit needed). - ---- - -### Task 1: `join_on_ts` helper + `JoinedRow` + RED tests (aura-engine) - -**Files:** -- Modify: `crates/aura-engine/src/report.rs` -- Modify: `crates/aura-engine/src/lib.rs:64` - -- [ ] **Step 1: Write the two failing tests** - -Append these two functions at the end of the `mod tests` block in -`crates/aura-engine/src/report.rs`, immediately before the module's closing brace -(after the last existing test `runreport_serde_round_trips`, ~:404). `Scalar`, -`Timestamp` are already in scope via the test module's `use super::*;`. - -```rust - #[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)])); - } -``` - -- [ ] **Step 2: Run the tests to verify they fail (RED)** - -Run: `cargo test -p aura-engine join_on_ts` -Expected: FAIL — compile error `cannot find function \`join_on_ts\` in this scope` -(and `cannot find type \`JoinedRow\``), because neither exists yet. - -- [ ] **Step 3: Add the `HashMap` import** - -In `crates/aura-engine/src/report.rs`, immediately after the existing top-of-file -import `use aura_core::{Scalar, ScalarKind, Timestamp};` (:11), add: - -```rust -use std::collections::HashMap; -``` - -- [ ] **Step 4: Write `JoinedRow` + `join_on_ts`** - -Insert between the end of `f64_field` (:144) and the `#[cfg(test)]` attribute (:146): - -```rust -/// 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() -} -``` - -- [ ] **Step 5: Re-export from `lib.rs`** - -In `crates/aura-engine/src/lib.rs:64`, replace: - -```rust -pub use report::{f64_field, summarize, RunManifest, RunMetrics, RunReport}; -``` - -with: - -```rust -pub use report::{f64_field, join_on_ts, summarize, JoinedRow, RunManifest, RunMetrics, RunReport}; -``` - -- [ ] **Step 6: Run the tests to verify they pass (GREEN)** - -Run: `cargo test -p aura-engine join_on_ts` -Expected: PASS — 2 tests pass (`join_on_ts_aligns_streams_of_different_cardinality`, -`join_on_ts_drops_side_rows_absent_from_spine`). - -- [ ] **Step 7: Verify the whole engine crate still builds + tests green** - -Run: `cargo test -p aura-engine` -Expected: PASS — all `report::tests` plus the rest of the crate's tests pass; no -warnings about the new `HashMap` import. - ---- - -### Task 2: Refactor `drain_trace` onto `join_on_ts` (aura-ingest) - -**Files:** -- Modify: `crates/aura-ingest/examples/shared/breakout_real.rs:42-45` (use block), - `:354-381` (`drain_trace` body) - -- [ ] **Step 1: Import `join_on_ts`** - -In `crates/aura-ingest/examples/shared/breakout_real.rs`, the `aura_engine` use -block (:42-45) currently reads: - -```rust -use aura_engine::{ - summarize, Composite, Edge, FlatGraph, GraphBuilder, Harness, RunManifest, RunReport, - SourceSpec, Target, -}; -``` - -Add `join_on_ts` (alphabetical, before `summarize`): - -```rust -use aura_engine::{ - join_on_ts, summarize, Composite, Edge, FlatGraph, GraphBuilder, Harness, RunManifest, - RunReport, SourceSpec, Target, -}; -``` - -(Import only `join_on_ts`, not `JoinedRow` — the closure infers the type, so naming -`JoinedRow` would be an unused import.) - -- [ ] **Step 2: Replace the `drain_trace` body** - -Replace the entire current `drain_trace` function (:354-381): - -```rust -pub fn drain_trace(taps: &Taps) -> Vec { - use std::collections::HashMap; - - let equity: Vec<(Timestamp, Vec)> = taps.equity.try_iter().collect(); - let held: Vec<(Timestamp, Vec)> = taps.held.try_iter().collect(); - let bars: Vec<(Timestamp, Vec)> = taps.bars.try_iter().collect(); - let breakout: Vec<(Timestamp, Vec)> = taps.breakout.try_iter().collect(); - - // held shares the equity spine (same node-firing cadence); align by index is - // safe between those two, but we still fuse the side taps by ts. - let held_by_ts: HashMap = - held.iter().map(|(t, r)| (t.0, as_f64(&r[0]))).collect(); - let bars_by_ts: HashMap = - bars.iter().map(|(t, r)| (t.0, as_i64(&r[0]))).collect(); - let breakout_by_ts: HashMap = - breakout.iter().map(|(t, r)| (t.0, as_bool(&r[0]))).collect(); - - equity - .iter() - .map(|(ts, row)| BarTrace { - ts: *ts, - equity: as_f64(&row[0]), - held: held_by_ts.get(&ts.0).copied().unwrap_or(0.0), - bars_since_open: bars_by_ts.get(&ts.0).copied().unwrap_or(-1), - breakout: breakout_by_ts.get(&ts.0).copied().unwrap_or(false), - }) - .collect() -} -``` - -with the join-based body (the function-local `use std::collections::HashMap;` is -dropped; the three `_by_ts` maps are gone; the caller-side defaults are identical): - -```rust -pub fn drain_trace(taps: &Taps) -> Vec { - let equity: Vec<(Timestamp, Vec)> = taps.equity.try_iter().collect(); - let held: Vec<(Timestamp, Vec)> = taps.held.try_iter().collect(); - let bars: Vec<(Timestamp, Vec)> = taps.bars.try_iter().collect(); - let breakout: Vec<(Timestamp, Vec)> = taps.breakout.try_iter().collect(); - - join_on_ts(&equity, &[&held, &bars, &breakout]) - .into_iter() - .map(|j| BarTrace { - ts: j.ts, - equity: as_f64(&j.spine[0]), - held: j.sides[0].as_ref().map_or(0.0, |r| as_f64(&r[0])), - bars_since_open: j.sides[1].as_ref().map_or(-1, |r| as_i64(&r[0])), - breakout: j.sides[2].as_ref().map_or(false, |r| as_bool(&r[0])), - }) - .collect() -} -``` - -- [ ] **Step 3: Build aura-ingest (example + tests compile)** - -Run: `cargo build -p aura-ingest --all-targets` -Expected: clean build, no warnings (the dead function-local `HashMap` import is -gone; `Timestamp`/`Scalar` are still used by the `try_iter().collect()` turbofish). - -- [ ] **Step 4: Acceptance grep — the hand-rolled join is gone** - -Run: `git grep -nE '_by_ts' -- crates/aura-ingest/examples/shared/breakout_real.rs` -Expected: no output (exit status 1) — no `_by_ts` HashMap construction or lookup -remains. - -- [ ] **Step 5: Behaviour-preservation — the gated GER40 tests stay green** - -Run: `cargo test -p aura-ingest` -Expected: PASS — `ger40_breakout_real_bars_run_is_deterministic`, -`composite_matches_flatgraph_bit_identical` (ger40_breakout_blueprint), and the -`open_ohlc_seam` tests all pass over the real archive (present at -`/mnt/tickdata/Pepperstone/GER40_2024_09.m1`); the refactored `drain_trace` yields a -byte-identical `RunReport`. (If the archive were absent these would skip — but it is -present, so they execute.) - ---- - -### Task 3: `Recorder` cadence/join doc note (aura-std) - -**Files:** -- Modify: `crates/aura-std/src/recorder.rs:13-16` - -- [ ] **Step 1: Append the prose note to the `Recorder` struct doc** - -In `crates/aura-std/src/recorder.rs`, the `Recorder` struct doc (:13-16) currently -reads: - -```rust -/// A recording sink over `kinds.len()` input columns. Each fired cycle it reads -/// the newest value of every column and sends the row to `tx`; it returns `None` -/// (records, forwards nothing) and `None` during warm-up until all columns have a -/// value. -``` - -Replace it with (the original four lines, then a new paragraph): - -```rust -/// A recording sink over `kinds.len()` input columns. Each fired cycle it reads -/// the newest value of every column and sends the row to `tx`; it returns `None` -/// (records, forwards nothing) and `None` during warm-up until all columns have a -/// value. -/// -/// Multiple taps fire at their own node's cadence, so their recorded streams have -/// different lengths and different firing instants. To fuse several taps into one -/// trace, join on the recorded timestamp — never by positional index, which -/// misaligns the columns or panics. The engine's report layer ships a `join_on_ts` -/// helper (in `aura-engine`) for exactly this; it is named in prose rather than -/// linked because `aura-std` does not depend on `aura-engine`. -``` - -- [ ] **Step 2: Verify the doc builds without warnings (no broken link)** - -Run: `cargo doc -p aura-std --no-deps 2>&1` -Expected: builds clean — no `unresolved link` / `intra-doc link` warning (the note -is plain prose, `join_on_ts` is not a `[...]` link). - ---- - -### Final gate (run after all tasks) - -- [ ] **Step 1: Full workspace build** - -Run: `cargo build --workspace --all-targets` -Expected: clean. - -- [ ] **Step 2: Full workspace test** - -Run: `cargo test --workspace` -Expected: PASS — the two new `join_on_ts` tests plus every existing test -(the gated GER40 tests run against the present archive). - -- [ ] **Step 3: Clippy** - -Run: `cargo clippy --workspace --all-targets -- -D warnings` -Expected: clean — no warnings (notably no unused-import warning for the new -`HashMap` in report.rs or the `join_on_ts` import in breakout_real.rs). diff --git a/docs/specs/0053-multi-tap-ts-join.md b/docs/specs/0053-multi-tap-ts-join.md deleted file mode 100644 index f7dbf75..0000000 --- a/docs/specs/0053-multi-tap-ts-join.md +++ /dev/null @@ -1,257 +0,0 @@ -# Multi-tap trace join on timestamp — Design Spec - -**Date:** 2026-06-18 -**Status:** Draft — awaiting user spec review -**Authors:** orchestrator + Claude - -## Goal - -Ship a reusable, tested helper that fuses several recording-sink (`Recorder`) -tap streams into one per-row trace **by their recorded timestamp**, so the next -consumer that taps multiple sinks does not rediscover — via a panic — that the -taps fire at different cardinalities and cannot be zipped by positional index -(#93). - -Today the only multi-tap consumer, the GER40 breakout demonstration, hand-rolls -this join inside `drain_trace` -(`crates/aura-ingest/examples/shared/breakout_real.rs`). The four sinks do not -fire on the same cadence — the `breakout` (Gt) tap is one row shorter because -`Delay(1)` is cold on the first bar, and the `bars_since_open` (Session) tap is -silent before a session open — so the first cut zipped by index and panicked. -This cycle lifts the timestamp-keyed join into a generic engine helper, refactors -`drain_trace` onto it, and documents the cadence rule on the `Recorder` API. - -Non-blocking. `closes #93`. - -## Architecture - -The helper is a **post-run pure reduction over recorded sink streams**, exactly -the family `aura-engine/src/report.rs` already houses (`summarize`, `f64_field`). -That module's own doc frames the layer: "a node cannot reduce end-of-run … so the -World drains its recording sinks after `Harness::run` and folds them here." A -timestamp join is a sibling reduction, so it lives there and is exported from -`crates/aura-engine/src/lib.rs` alongside `summarize` / `f64_field`. - -Invariant fit: - -- **C8 / C18** — a sink is the only observable surface. The helper consumes only - what sinks recorded (`(Timestamp, Vec)` streams), never engine - internals, and reduces them after the run. -- **C3** — there is one merge, at ingestion only; no in-graph as-of join. This - join runs **post-run, over already-recorded output**, not inside the graph, so - it does not reintroduce an in-graph merge. -- **C1** — a backtest reaches a unique state per tick, so each cycle has a unique - timestamp and a sink fires at most once per timestamp. Per-stream - timestamp-uniqueness is therefore a documented precondition, not a runtime - check. - -The join is **spine-anchored**: one designated stream (the densest — the -strategy's primary recorded output) defines the row set, and the other streams -are looked up against it. This matches the domain shape — the spine is the -per-bar equity/exposure series, the side taps annotate *why* each bar did or did -not act. - -The default-on-miss semantics are **Option-per-side**: the helper reports -*presence* (`Some(row)` where a side fired at a spine timestamp, `None` where it -did not), and the consumer interprets absence. The consumer-level defaults -(`held → 0.0`, `bars_since_open → -1`, `breakout → false`) are interpretations of -"this node did not fire" that only the consumer can justify; a generic engine -helper must not invent them. (Resolved fork; see the reconciliation comment on -issue #93.) - -## Concrete code shapes - -### User-facing program (the acceptance-criterion evidence) - -A fresh consumer with N taps of its own — the program the helper must make work: - -```rust -use aura_engine::{join_on_ts, JoinedRow}; - -// Drain each tap's recording channel into its own (ts, row) Vec. The streams -// have DIFFERENT lengths — that is the whole point. -let equity: Vec<(Timestamp, Vec)> = rx_equity.try_iter().collect(); // spine -let held: Vec<(Timestamp, Vec)> = rx_held.try_iter().collect(); -let signal: Vec<(Timestamp, Vec)> = rx_signal.try_iter().collect(); - -// One JoinedRow per spine entry; each side is Some(row) / None at that ts. -for j in join_on_ts(&equity, &[&held, &signal]) { - let eq = as_f64(&j.spine[0]); - let h = j.sides[0].as_ref().map_or(0.0, |r| as_f64(&r[0])); // default is MINE - let sig = j.sides[1].as_ref().map_or(false, |r| as_bool(&r[0])); // not the engine's - // ... fold (j.ts, eq, h, sig) into the consumer's trace row -} -``` - -And the existing consumer, `drain_trace`, refactored onto it (its `BarTrace` -mapping becomes a thin layer; defaults stay caller-side and unchanged): - -```rust -// crates/aura-ingest/examples/shared/breakout_real.rs -pub fn drain_trace(taps: &Taps) -> Vec { - let equity: Vec<(Timestamp, Vec)> = taps.equity.try_iter().collect(); - let held: Vec<(Timestamp, Vec)> = taps.held.try_iter().collect(); - let bars: Vec<(Timestamp, Vec)> = taps.bars.try_iter().collect(); - let breakout: Vec<(Timestamp, Vec)> = taps.breakout.try_iter().collect(); - - join_on_ts(&equity, &[&held, &bars, &breakout]) - .into_iter() - .map(|j| BarTrace { - ts: j.ts, - equity: as_f64(&j.spine[0]), - held: j.sides[0].as_ref().map_or(0.0, |r| as_f64(&r[0])), - bars_since_open: j.sides[1].as_ref().map_or(-1, |r| as_i64(&r[0])), - breakout: j.sides[2].as_ref().map_or(false, |r| as_bool(&r[0])), - }) - .collect() -} -``` - -### Implementation shape (secondary — before → after) - -**New in `aura-engine/src/report.rs`** (there is no "before"): - -```rust -/// 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. -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() -} -``` - -**Changed in `aura-engine/src/lib.rs`** — extend the `report` re-export: - -```rust -// before -pub use report::{f64_field, summarize, RunManifest, RunMetrics, RunReport}; -// after -pub use report::{f64_field, join_on_ts, summarize, JoinedRow, RunManifest, RunMetrics, RunReport}; -``` - -**Changed in `crates/aura-std/src/recorder.rs`** — a prose doc note on the -`Recorder` type. It must be plain prose, NOT a rustdoc intra-doc link: -`aura-std` is a dependency of `aura-engine`, so it cannot link to -`aura_engine::join_on_ts` without a dependency cycle. Shape: - -```rust -/// ... existing Recorder doc ... -/// -/// Multiple taps fire at their own node's cadence, so their recorded streams -/// have different lengths and different firing instants. To fuse several taps -/// into one trace, join on the recorded timestamp — never by positional index -/// (positional zip misaligns or panics). The engine ships a `join_on_ts` helper -/// (in `aura-engine`'s report layer) for exactly this. -``` - -## Components - -1. **`JoinedRow`** (new, `report.rs`, public) — `{ ts: Timestamp, spine: - Vec, sides: Vec>> }`. -2. **`join_on_ts`** (new, `report.rs`, public) — the spine-anchored, - Option-per-side join; re-exported from `lib.rs`. -3. **`drain_trace`** (refactor, `breakout_real.rs`) — its hand-rolled HashMap - join is replaced by a `join_on_ts` call; the `BarTrace` mapping and its - caller-side defaults are unchanged in behaviour. -4. **`Recorder` doc note** (`recorder.rs`) — the cadence/join rule, as prose. - -No new types beyond `JoinedRow`; `Scalar` / `Timestamp` are the building blocks. - -## Data flow - -`Harness::run` finishes → the consumer drains each tap's `mpsc::Receiver` into a -`Vec<(Timestamp, Vec)>`. The streams have different lengths. The consumer -picks the densest stream as the spine and passes the rest as `sides`. `join_on_ts` -indexes each side once into a `HashMap>`, then walks the spine -in order, emitting one `JoinedRow` per spine entry with each side looked up by -`ts.0` → `Some(clone)` / `None`. The consumer maps each `JoinedRow` to its own -trace row, supplying its own default for each `None`. - -## Error handling - -The helper never panics and performs no kind checking — it returns side rows -whole, so column/kind interpretation (`as_f64`, `as_i64`, `as_bool`) stays in the -consumer exactly as today. Edge cases, all by construction rather than by guard: - -- **Side timestamp absent from the spine** → that side row is dropped (the spine - defines the row set). Documented, spine-anchored semantics. -- **Duplicate timestamp within one stream** → last-write-wins (the `HashMap` - collect keeps the last). Cannot occur for a deterministic run (C1); documented - as a precondition rather than enforced. -- **Empty spine** → empty output. -- **A side stream longer than the spine** → its extra rows are simply never - looked up (dropped, as above). - -## Testing strategy - -RED unit tests in `report.rs`'s test module (they reference `join_on_ts` / -`JoinedRow`, which do not yet exist → compile-fail = RED, the executable spec for -the helper): - -1. **`join_on_ts_aligns_streams_of_different_cardinality`** — the #93 shape: a - spine firing every bar; side A one row shorter (skips the first timestamp, - like cold `Delay(1)`); side B firing only on a subset (like a Session filter). - Assert: output length == spine length; `Some`/`None` per side at the right - timestamps; the `Some` values are the side rows recorded at that timestamp - (the alignment a positional zip gets wrong). -2. **`join_on_ts_drops_side_rows_absent_from_spine`** — a side row at a timestamp - not present in the spine is absent from the output; the in-spine row is still - joined. - -Behaviour-preservation of the refactor is ratified end-to-end by the **existing -gated GER40 tests** (`crates/aura-ingest/tests/ger40_breakout_real.rs`, -`crates/aura-ingest/tests/ger40_breakout_blueprint.rs`): they drain the same four -taps through `drain_trace` and fold the result into a `RunReport` over real data, -so the refactored `drain_trace` must keep them green (byte-identical trace). No -new GER40 fixture is added. - -`cargo test --workspace`, `cargo build --workspace`, and `cargo clippy ---workspace --all-targets -- -D warnings` are the gate. - -## Acceptance criteria - -- `aura_engine::join_on_ts` and `aura_engine::JoinedRow` exist and are - re-exported from `lib.rs`. -- The two RED unit tests above exist and pass. -- `drain_trace` is refactored onto `join_on_ts`; its hand-rolled per-tap - `HashMap` join is gone (acceptance grep: no `_by_ts` HashMap construction left - in `breakout_real.rs`). -- The existing gated GER40 tests pass unchanged (behaviour-preserved trace). -- `Recorder`'s doc carries the cadence/join note as prose (no cross-crate - rustdoc intra-doc link). -- `cargo build --workspace`, `cargo test --workspace`, and `cargo clippy - --workspace --all-targets -- -D warnings` are clean.