Brummel/Aura
1
0
Fork 0
Code Issues 33 Pull Requests Actions Packages Projects Releases Wiki Activity

chore: cycle 0053 close — audit clean, retire spec + plan

Browse Source 
Cycle 0053 (multi-tap trace join on timestamp, #93) closes drift-clean.

Audit (architect drift review over 066638e..HEAD against docs/design + the
CLAUDE.md domain invariants):
- C8/C18 honoured: join_on_ts is a post-run reduction over recorded
  (Timestamp, Vec<Scalar>) sink streams, a true sibling of summarize/f64_field
  in the report layer — correct architectural home.
- C3 not violated: the join runs post-run over already-recorded output, not
  in-graph; no as-of join smuggled into the DAG. C1 ts-uniqueness is a
  documented precondition, not a runtime check — consistent with how the
  engine leaves wiring-time invariants to contract.
- Option-per-side keeps the engine honest (reports presence; the consumer owns
  the 0.0/-1/false defaults). The hand-rolled *_by_ts HashMaps are fully gone;
  the two RED unit tests pin the #93 cardinality-misalignment + spine-drop.
- No drift, no debt, no ledger entry needs updating. Recommendation: carry on.

Regression gate: no dedicated regression scripts configured (optional per the
conventions); build/test/clippy --workspace --all-targets is the gate and ran
green — the gated GER40 byte-identity tests executed against the local archive
(not skipped), proving drain_trace's trace is byte-preserved across the
refactor.

Per the artifact lifecycle (committed while the cycle is live, git rm at cycle
close), the ephemeral spec and plan are retired here; their durable rationale
lives in the implementation commits 74324d1 / 35c5adc and the git history.
This commit is contained in:
Brummel
2026-06-18 15:04:42 +02:00
parent 35c5adc6f3
commit 7deb38edd8
2 changed files with 0 additions and 642 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/plans/0053-multi-tap-ts-join.md
-385
View File
@@ -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<_>>(),
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<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:
```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<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):
```rust
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:
```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).
docs/specs/0053-multi-tap-ts-join.md
-257
View File
@@ -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<Scalar>)` 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<Scalar>)> = rx_equity.try_iter().collect(); // spine
let held: Vec<(Timestamp, Vec<Scalar>)> = rx_held.try_iter().collect();
let signal: Vec<(Timestamp, Vec<Scalar>)> = 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<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()
}
```
### 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<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()
}
```
**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<Scalar>, sides: Vec<Option<Vec<Scalar>>> }`.
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<Scalar>)>`. 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<i64, &Vec<Scalar>>`, 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.