refactor(aura-ingest): fuse drain_trace taps via join_on_ts; doc the cadence rule

drain_trace now joins its four recording-sink taps through aura-engine's
join_on_ts instead of a hand-rolled per-tap HashMap, resolving the #93
zip-by-index panic at its source: the four taps fire at different
cardinalities (breakout one bar short via cold Delay(1); bars_since_open
filtered by Session), so a positional zip misaligns or panics. The BarTrace
mapping stays a thin layer with its caller-side defaults unchanged
(held->0.0, bars_since_open->-1, breakout->false) — the engine reports
presence, the consumer interprets absence. Behaviour is byte-preserved: the
gated GER40 tests (ger40_breakout_real, ger40_breakout_blueprint,
open_ohlc_seam) stay green against the real archive.

The Recorder doc gains a prose note on the multi-tap cadence rule (join on
ts, never by index), pointing at join_on_ts by name rather than rustdoc link
since aura-std does not depend on aura-engine.

Verified: cargo build/test/clippy --workspace --all-targets clean; the
_by_ts hand-rolled join is gone (acceptance grep empty).

closes #93

crates/aura-ingest/examples/shared/breakout_real.rs

@@ -40,8 +40,8 @@ use std::sync::mpsc;
 
 use aura_core::{NodeSchema, Scalar, ScalarKind, Timestamp};
 use aura_engine::{
-    summarize, Composite, Edge, FlatGraph, GraphBuilder, Harness, RunManifest, RunReport,
+    join_on_ts, summarize, Composite, Edge, FlatGraph, GraphBuilder, Harness, RunManifest,
-    SourceSpec, Target,
+    RunReport, SourceSpec, Target,
 };
 use aura_std::{And, Delay, EqConst, Gt, Latch, Recorder, Resample, Session, SimBroker};
 use chrono::TimeZone;
@@ -352,30 +352,19 @@ pub struct BarTrace {
 /// (the sinks fire on different cardinalities — see [`BarTrace`]). Reads only
 /// recorded sink output (C8/C18) — never engine internals.
 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
+    join_on_ts(&equity, &[&held, &bars, &breakout])
-    // safe between those two, but we still fuse the side taps by ts.
+        .into_iter()
-    let held_by_ts: HashMap<i64, f64> =
+        .map(|j| BarTrace {
-        held.iter().map(|(t, r)| (t.0, as_f64(&r[0]))).collect();
+            ts: j.ts,
-    let bars_by_ts: HashMap<i64, i64> =
+            equity: as_f64(&j.spine[0]),
-        bars.iter().map(|(t, r)| (t.0, as_i64(&r[0]))).collect();
+            held: j.sides[0].as_ref().map_or(0.0, |r| as_f64(&r[0])),
-    let breakout_by_ts: HashMap<i64, bool> =
+            bars_since_open: j.sides[1].as_ref().map_or(-1, |r| as_i64(&r[0])),
-        breakout.iter().map(|(t, r)| (t.0, as_bool(&r[0]))).collect();
+            breakout: j.sides[2].as_ref().is_some_and(|r| as_bool(&r[0])),
-
-    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()
 }

crates/aura-std/src/recorder.rs

@@ -14,6 +14,13 @@ use std::sync::mpsc::Sender;
 /// 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`.
 pub struct Recorder {
     kinds: Vec<ScalarKind>,
     tx: Sender<(Timestamp, Vec<Scalar>)>,