From a1625859814f95dbceba786aaf57298a6bb1e57c Mon Sep 17 00:00:00 2001 From: Brummel Date: Thu, 18 Jun 2026 14:47:11 +0200 Subject: [PATCH] plan: 0053 multi-tap ts-join Bite-sized execution plan for spec 0053: Task 1 the join_on_ts helper + JoinedRow + two RED unit tests in aura-engine report.rs (re-exported from lib.rs); Task 2 the behaviour-preserving drain_trace refactor onto the helper; Task 3 the Recorder prose cadence/join note. Final gate is workspace build/test/clippy. refs #93 --- docs/plans/0053-multi-tap-ts-join.md | 385 +++++++++++++++++++++++++++ 1 file changed, 385 insertions(+) create mode 100644 docs/plans/0053-multi-tap-ts-join.md diff --git a/docs/plans/0053-multi-tap-ts-join.md b/docs/plans/0053-multi-tap-ts-join.md new file mode 100644 index 0000000..a9258ed --- /dev/null +++ b/docs/plans/0053-multi-tap-ts-join.md @@ -0,0 +1,385 @@ +# 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).