Aura/docs/plans/0053-multi-tap-ts-join.md

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::*;.

    #[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)]));
    }

• 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(andcannot 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:

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):

/// 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()
}

• Step 5: Re-export from lib.rs

In crates/aura-engine/src/lib.rs:64, replace:

pub use report::{f64_field, summarize, RunManifest, RunMetrics, RunReport};

with:

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:

use aura_engine::{
    summarize, Composite, Edge, FlatGraph, GraphBuilder, Harness, RunManifest, RunReport,
    SourceSpec, Target,
};

Add join_on_ts (alphabetical, before summarize):

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):

pub fn drain_trace(taps: &Taps) -> Vec<BarTrace> {
    use std::collections::HashMap;

    let equity: Vec<(Timestamp, Vec<Scalar>)> = taps.equity.try_iter().collect();
    let held: Vec<(Timestamp, Vec<Scalar>)> = taps.held.try_iter().collect();
    let bars: Vec<(Timestamp, Vec<Scalar>)> = taps.bars.try_iter().collect();
    let breakout: Vec<(Timestamp, Vec<Scalar>)> = 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<i64, f64> =
        held.iter().map(|(t, r)| (t.0, as_f64(&r[0]))).collect();
    let bars_by_ts: HashMap<i64, i64> =
        bars.iter().map(|(t, r)| (t.0, as_i64(&r[0]))).collect();
    let breakout_by_ts: HashMap<i64, bool> =
        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):

pub fn drain_trace(taps: &Taps) -> Vec<BarTrace> {
    let equity: Vec<(Timestamp, Vec<Scalar>)> = taps.equity.try_iter().collect();
    let held: Vec<(Timestamp, Vec<Scalar>)> = taps.held.try_iter().collect();
    let bars: Vec<(Timestamp, Vec<Scalar>)> = taps.bars.try_iter().collect();
    let breakout: Vec<(Timestamp, Vec<Scalar>)> = 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:

/// 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):

/// 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).