spec: 0053 multi-tap ts-join
Ratified design for a reusable, spine-anchored join_on_ts helper in aura-engine that fuses multi-tap Recorder traces on the recorded timestamp (Option-per-side: the engine reports presence, the consumer interprets absence). Refactors drain_trace onto it and documents the cadence rule on the Recorder API. Resolves the #93 zip-by-index panic. refs #93
This commit is contained in:
@@ -0,0 +1,257 @@
|
||||
# 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.
|
||||
Reference in New Issue
Block a user