Aura/crates/aura-engine/src/report.rs

//! Run summary metrics + the reproducible run manifest (C18 / C12): the
//! `(manifest, metrics)` pair a run produces "from day one". The metrics are a
//! **post-run pure reduction** over a run's recorded streams — a node cannot
//! reduce end-of-run (C8 caps a node at one record per `eval`, with no terminal
//! `eval`), so the World drains its recording sinks after [`Harness::run`](crate::Harness::run)
//! and folds them here. Output is canonical JSON (C14): the schema is tiny,
//! closed, and flat. `to_json` renders via serde (the report types derive it,
//! cycle 0029) — the same encoder the run registry uses, so a record's stdout
//! and on-disk shapes coincide.

use aura_core::{Scalar, ScalarKind, Timestamp};
use std::collections::HashMap;

// The backtest reductions (R-metrics, the position-event table, `summarize`)
// moved to `aura-backtest::metrics` (C28 phase 2, #291/#292 — the engine's
// production edge to aura-backtest is cut, D4); the domain-free hurdle math +
// selection provenance stays re-imported here from `aura-analysis` so
// `report::`'s namespace — and therefore `aura-engine`'s `lib.rs` re-export and
// every `crate::report::X` reference — resolves unchanged for those names
// (C18 byte-identity).
pub use aura_analysis::{expected_max_of_normals, inv_norm_cdf, FamilySelection, SelectionMode};

/// The reproducible run descriptor (C18). **Caller-supplied**: the engine
/// cannot introspect a git commit, an RNG seed, or a broker label — the World
/// that bootstraps and runs the harness fills these in.
#[derive(Clone, Debug, PartialEq, serde::Serialize, serde::Deserialize)]
pub struct RunManifest {
    /// Node/engine identity: the git commit of the frozen artifact (C18 —
    /// commit = identity; the frozen bot *is* a commit).
    pub commit: String,
    /// The bound tuning params as ordered `name -> value` pairs. Each value is a
    /// self-describing [`Scalar`], so the param's kind (an `i64` length vs an
    /// `f64` scale) survives into the record instead of collapsing to `f64`.
    pub params: Vec<(String, Scalar)>,
    /// The bound params this run did NOT vary — the wrap-prefixed
    /// `bound_param_space()` of the signal, EXCLUDING any bound param an axis
    /// reopened (those flow through `params` instead, #246). Complements
    /// `params` ("what varied") with "what was held": the two are disjoint by
    /// construction. Empty for every pre-#249 record. One-directional serde
    /// widening (C14/C18), identical idiom to `selection`/`instrument`/
    /// `topology_hash` — except a `Vec`, not an `Option`, so it always
    /// serializes (mirroring `params`), never skipped when empty.
    #[serde(default)]
    pub defaults: Vec<(String, Scalar)>,
    /// The data-window: inclusive `(from, to)` epoch-ns bounds (C12).
    pub window: (Timestamp, Timestamp),
    /// The RNG seed (C12 seed-as-input). `0` for a seed-free synthetic run.
    pub seed: u64,
    /// The broker profile label, e.g. `"sim-optimal(pip_size=0.0001)"`.
    pub broker: String,
    /// Selection provenance, present only on a sweep/walk-forward winner; a
    /// standalone run and every pre-0076 line read back as `None`. One-directional
    /// serde widening (C14/C18), identical idiom to `RunMetrics.r`.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub selection: Option<FamilySelection>,
    /// The instrument this run evaluated, when it is a real-data run; `None` for a
    /// synthetic run and for every pre-0078 line. The "instrument set as lineage"
    /// (C18) for a cross-instrument family is each member's stamped symbol. Same
    /// one-directional serde widening as `selection` (C14/C18).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub instrument: Option<String>,
    /// SHA256 (hex) of the canonical `blueprint_to_json` of the run's signal
    /// topology — the #158 reproducibility anchor. `None` for a run not built
    /// from a serialized blueprint and for every pre-0092 line. One-directional
    /// serde widening (Tier-1, #156), identical idiom to `selection` / `instrument`.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub topology_hash: Option<String>,
    /// Identity of the loaded project dylib, when the run used one (cycle
    /// 0102). Pre-0102 records read as `None`; `None` serializes as an absent
    /// key (the same one-directional widening as `instrument`).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub project: Option<ProjectProvenance>,
}

/// Provenance of the project cdylib a run was resolved through (C16/C18).
/// `namespace`/`dylib_sha256` are present only when a node crate is loaded
/// (native tier); a data-only project stamps only its commit.
#[derive(Clone, Debug, PartialEq, serde::Serialize, serde::Deserialize)]
pub struct ProjectProvenance {
    /// The project's vocabulary namespace (from the descriptor).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub namespace: Option<String>,
    /// SHA-256 (hex) of the loaded dylib file's bytes.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub dylib_sha256: Option<String>,
    /// The project repo's HEAD (+ `-dirty`), when derivable.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub commit: Option<String>,
}

/// A run's full structured result: the descriptor plus the metric payload `M`
/// it reproduces. Metric-agnostic (C28): the engine never names a concrete
/// metric type; the backtest layer instantiates `M` (see `aura-backtest`). The
/// durable run record of C18 ("stores manifests + metrics, re-derives full
/// results on demand").
#[derive(Clone, Debug, PartialEq, serde::Serialize, serde::Deserialize)]
pub struct RunReport<M> {
    pub manifest: RunManifest,
    pub metrics: M,
}

impl<M: serde::Serialize> RunReport<M> {
    /// Render the canonical, machine-readable JSON (C14) via serde — the same
    /// encoder the run registry uses on disk, so a record's stdout shape and its
    /// `runs.jsonl` shape are byte-identical. `params` is an array of
    /// `[name, value]` pairs where `value` is a self-describing tagged scalar
    /// (serde's externally-tagged enum: `{"I64": 10}` for a length, `{"F64": 2.5}`
    /// for a scale). Consumers parse the tagged object, never a bare number.
    pub fn to_json(&self) -> String {
        serde_json::to_string(self).expect("a finite RunReport always serializes")
    }
}

/// A measurement run's stdout/record shape (C28 phase 3): the run descriptor plus
/// the declared tap names, mirroring the persisted `index.json`. A measurement run
/// has no broker → no equity/exposure → NO R `metrics` (the difference from
/// [`RunReport`]). The tapped series themselves persist through the trace store.
#[derive(Clone, Debug, PartialEq, serde::Serialize, serde::Deserialize)]
pub struct MeasurementReport {
    pub manifest: RunManifest,
    pub taps: Vec<String>,
}

impl MeasurementReport {
    /// Canonical machine-readable JSON via serde, same encoder as [`RunReport`].
    pub fn to_json(&self) -> String {
        serde_json::to_string(self).expect("a finite MeasurementReport always serializes")
    }
}

/// Bridge a recording sink's recorded `(ts, row)` stream to a metric reduction
/// (e.g. `aura-backtest`'s `summarize`): extract one `f64` field of each row
/// into `(ts, f64)` samples. Panics if a
/// row has no such field or the field is not an `f64` scalar — a wiring bug (a
/// sink's declared kinds are fixed at bootstrap, so a correctly-wired
/// equity/exposure sink always yields `f64` at field 0), surfaced like the
/// engine's other "checked at wiring" contract violations rather than silently
/// dropped.
pub fn f64_field(rows: &[(Timestamp, Vec<Scalar>)], field: usize) -> Vec<(Timestamp, f64)> {
    rows.iter()
        .map(|(ts, row)| {
            let Some(&scalar) = row.get(field) else {
                panic!("f64_field: row has no field {field} (row width {})", row.len());
            };
            if scalar.kind() != ScalarKind::F64 {
                panic!("f64_field: field {field} is not an f64 scalar: {scalar:?}");
            }
            (*ts, scalar.as_f64())
        })
        .collect()
}

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

/// One drained recorder tap in columnar (SoA) form: parallel arrays, one per
/// recorded column, plus the shared recorded-timestamp axis. Chart-ready (a column
/// is a series of numbers) and kind-tagged (C7) — values are coerced to f64 for
/// plotting; the base type survives in `kinds`. Pure: identical rows always encode
/// to the same value (C1).
#[derive(Clone, Debug, PartialEq, serde::Serialize, serde::Deserialize)]
pub struct ColumnarTrace {
    pub tap: String,
    pub kinds: Vec<String>,
    pub ts: Vec<i64>,
    pub columns: Vec<Vec<f64>>,
}

impl ColumnarTrace {
    /// Transpose a drained tap's `(ts, row)` pairs into columns. An empty `rows`
    /// yields empty `ts` and one empty column per kind. Each cell is coerced to f64:
    /// f64 as-is, i64 as f64, bool 1.0/0.0, timestamp epoch as f64. Panics if any
    /// row's width disagrees with `kinds.len()` — a wiring bug (a sink's column
    /// count is fixed at bootstrap), surfaced as a named panic like
    /// [`f64_field`] rather than a bare index-out-of-bounds (wider) or silent
    /// ragged columns (narrower).
    pub fn from_rows(tap: &str, kinds: &[ScalarKind], rows: &[(Timestamp, Vec<Scalar>)]) -> Self {
        let ts: Vec<i64> = rows.iter().map(|(t, _)| t.0).collect();
        let mut columns: Vec<Vec<f64>> = vec![Vec::with_capacity(rows.len()); kinds.len()];
        for (_, row) in rows {
            if row.len() != kinds.len() {
                panic!(
                    "from_rows: row width {} disagrees with kinds.len() {} (tap {tap:?})",
                    row.len(),
                    kinds.len(),
                );
            }
            for (c, scalar) in row.iter().enumerate() {
                columns[c].push(scalar_to_f64(*scalar));
            }
        }
        ColumnarTrace {
            tap: tap.to_string(),
            kinds: kinds.iter().map(kind_tag).collect(),
            ts,
            columns,
        }
    }

    /// Inverse for the serve/align path: rebuild `(ts, row)` pairs with each cell as
    /// `Scalar::f64(columns[c][r])` — uniformly f64 (the on-disk store is f64
    /// columns; the base type survives only in `kinds`). A true value round-trip for
    /// f64 taps; keeps the serve-side `as_f64` projection total.
    pub fn to_rows(&self) -> Vec<(Timestamp, Vec<Scalar>)> {
        self.ts
            .iter()
            .enumerate()
            .map(|(r, &t)| {
                let row = self.columns.iter().map(|col| Scalar::f64(col[r])).collect();
                (Timestamp(t), row)
            })
            .collect()
    }
}

/// The four-kind tag string for a column (the on-disk `kinds` form; `ScalarKind`
/// derives no serde, so tags are plain strings).
fn kind_tag(kind: &ScalarKind) -> String {
    match kind {
        ScalarKind::F64 => "F64",
        ScalarKind::I64 => "I64",
        ScalarKind::Bool => "Bool",
        ScalarKind::Timestamp => "Timestamp",
    }
    .to_string()
}

/// Coerce a recorded scalar to the f64 a chart plots: f64 as-is, i64 as f64,
/// bool 1.0/0.0, timestamp epoch as f64.
fn scalar_to_f64(s: Scalar) -> f64 {
    match s.kind() {
        ScalarKind::F64 => s.as_f64(),
        ScalarKind::I64 => s.as_i64() as f64,
        ScalarKind::Bool => {
            if s.as_bool() {
                1.0
            } else {
                0.0
            }
        }
        ScalarKind::Timestamp => s.as_ts().0 as f64,
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use aura_backtest::RunMetrics;

    #[test]
    fn runmanifest_without_selection_field_deserialises_to_none() {
        let json = r#"{"commit":"c","params":[],"window":[0,0],"seed":0,"broker":"b"}"#;
        let m: RunManifest = serde_json::from_str(json).expect("legacy manifest loads");
        assert!(m.selection.is_none());
    }

    #[test]
    fn runmanifest_without_selection_serialises_without_the_key() {
        let m = RunManifest {
            commit: "c".into(), params: vec![], defaults: vec![], window: (Timestamp(0), Timestamp(0)),
            seed: 0, broker: "b".into(), selection: None, instrument: None,
            topology_hash: None,
            project: None,
        };
        assert!(!serde_json::to_string(&m).unwrap().contains("selection"));
    }

    #[test]
    fn runmanifest_without_instrument_field_deserialises_to_none() {
        // A pre-0078 line carries no `instrument` key; serde `default` -> None.
        let json = r#"{"commit":"c","params":[],"window":[0,0],"seed":0,"broker":"b"}"#;
        let m: RunManifest = serde_json::from_str(json).expect("legacy manifest loads");
        assert!(m.instrument.is_none());
    }

    #[test]
    fn runmanifest_instrument_round_trips_and_omits_when_none() {
        let with = RunManifest {
            commit: "c".into(), params: vec![], defaults: vec![], window: (Timestamp(0), Timestamp(0)),
            seed: 0, broker: "b".into(), selection: None, instrument: Some("GER40".into()),
            topology_hash: None,
            project: None,
        };
        let json = serde_json::to_string(&with).unwrap();
        assert!(json.contains("\"instrument\":\"GER40\""), "json: {json}");
        let back: RunManifest = serde_json::from_str(&json).unwrap();
        assert_eq!(back.instrument, Some("GER40".to_string()));

        let without = RunManifest {
            commit: "c".into(), params: vec![], defaults: vec![], window: (Timestamp(0), Timestamp(0)),
            seed: 0, broker: "b".into(), selection: None, instrument: None,
            topology_hash: None,
            project: None,
        };
        // skip_serializing_if omits the key when None -> existing manifest bytes unchanged (C23).
        assert!(!serde_json::to_string(&without).unwrap().contains("instrument"));
    }

    #[test]
    fn runmanifest_project_field_round_trips_and_omits_when_none() {
        // A pre-0102 line (no "project" key) must load with project: None.
        let legacy = r#"{"commit":"c","params":[],"window":[0,1],"seed":7,"broker":"b"}"#;
        let m: RunManifest = serde_json::from_str(legacy).expect("legacy line loads");
        assert_eq!(m.project, None);
        // None must serialize with the key absent (byte-stability of old paths).
        let out = serde_json::to_string(&m).expect("serializes");
        assert!(!out.contains("\"project\""));
        // Some(..) must round-trip.
        let p = ProjectProvenance {
            namespace: Some("demo".into()),
            dylib_sha256: Some("ab".repeat(32)),
            commit: None,
        };
        let m2 = RunManifest { project: Some(p.clone()), ..m };
        let s = serde_json::to_string(&m2).expect("serializes");
        let back: RunManifest = serde_json::from_str(&s).expect("round-trips");
        assert_eq!(back.project, Some(p));
    }

    #[test]
    fn runmanifest_project_field_commit_only_omits_namespace_and_sha() {
        // #241: a data-only project (no node crate loaded) stamps a
        // commit-only ProjectProvenance; namespace/dylib_sha256 must stay
        // absent from the bytes (byte-stability of the native-tier shape)
        // and the commit-only value must still round-trip.
        let p = ProjectProvenance {
            namespace: None,
            dylib_sha256: None,
            commit: Some("deadbeef".into()),
        };
        let m = RunManifest {
            commit: "c".into(), params: vec![], defaults: vec![], window: (Timestamp(0), Timestamp(0)),
            seed: 0, broker: "b".into(), selection: None, instrument: None,
            topology_hash: None,
            project: Some(p.clone()),
        };
        let s = serde_json::to_string(&m).expect("serializes");
        assert!(!s.contains("\"namespace\""), "namespace omitted when None: {s}");
        assert!(!s.contains("\"dylib_sha256\""), "dylib_sha256 omitted when None: {s}");
        assert!(s.contains("\"commit\":\"deadbeef\""), "commit present: {s}");
        let back: RunManifest = serde_json::from_str(&s).expect("round-trips");
        assert_eq!(back.project, Some(p));
    }

    #[test]
    fn family_selection_round_trips_on_the_manifest() {
        let sel = FamilySelection {
            selection_metric: "sqn_normalized".into(), n_trials: 4, raw_winner_metric: 1.83,
            mode: SelectionMode::Argmax,
            deflated_score: Some(0.21), overfit_probability: Some(0.06),
            n_resamples: Some(1000), block_len: Some(5), seed: Some(42),
            neighbourhood_score: None, n_neighbours: None,
        };
        let m = RunManifest {
            commit: "c".into(), params: vec![], defaults: vec![], window: (Timestamp(0), Timestamp(0)),
            seed: 0, broker: "b".into(), selection: Some(sel.clone()), instrument: None,
            topology_hash: None,
            project: None,
        };
        let back: RunManifest = serde_json::from_str(&serde_json::to_string(&m).unwrap()).unwrap();
        assert_eq!(back.selection, Some(sel));
    }

    #[test]
    fn family_selection_plateau_shape_round_trips() {
        let sel = FamilySelection {
            selection_metric: "sqn_normalized".into(), n_trials: 16, raw_winner_metric: 1.42,
            mode: SelectionMode::PlateauMean,
            deflated_score: None, overfit_probability: None,
            n_resamples: None, block_len: None, seed: None,
            neighbourhood_score: Some(1.27), n_neighbours: Some(5),
        };
        let json = serde_json::to_string(&sel).unwrap();
        // plateau annotation present; deflation fields omitted (skip_serializing_if)
        assert!(json.contains("\"neighbourhood_score\":1.27"), "json: {json}");
        assert!(json.contains("\"n_neighbours\":5"), "json: {json}");
        assert!(!json.contains("deflated_score"), "deflation fields omitted under plateau: {json}");
        assert!(!json.contains("n_resamples"), "json: {json}");
        let back: FamilySelection = serde_json::from_str(&json).unwrap();
        assert_eq!(back, sel);
    }

    #[test]
    fn family_selection_loads_a_pre_0077_legacy_line() {
        // The pre-0077 on-disk shape (#144): the deflation fields as BARE scalars,
        // no plateau keys. `families.jsonl` / `runs.jsonl` are append-only, so a
        // winner stamped last cycle must still read back — the bare scalars land in
        // the `Some(...)` deflation Options, the absent plateau keys default to
        // `None` (the C14/C18 back-compat the struct's serde attrs claim).
        let legacy = r#"{"selection_metric":"sqn_normalized","n_trials":4,"raw_winner_metric":1.83,"deflated_score":0.21,"overfit_probability":0.06,"mode":"Argmax","n_resamples":1000,"block_len":5,"seed":42}"#;
        let sel: FamilySelection =
            serde_json::from_str(legacy).expect("a pre-0077 selection line still loads");
        assert_eq!(sel.mode, SelectionMode::Argmax);
        assert_eq!(sel.deflated_score, Some(0.21));
        assert_eq!(sel.overfit_probability, Some(0.06));
        assert_eq!(sel.n_resamples, Some(1000));
        assert_eq!(sel.block_len, Some(5));
        assert_eq!(sel.seed, Some(42));
        assert_eq!(sel.neighbourhood_score, None);
        assert_eq!(sel.n_neighbours, None);
    }

    #[test]
    fn f64_field_projects_the_named_field() {
        let rows = vec![
            (Timestamp(1), vec![Scalar::f64(1.5), Scalar::i64(9)]),
            (Timestamp(2), vec![Scalar::f64(2.5), Scalar::i64(8)]),
        ];
        assert_eq!(
            f64_field(&rows, 0),
            vec![(Timestamp(1), 1.5), (Timestamp(2), 2.5)],
        );
    }

    #[test]
    #[should_panic(expected = "not an f64 scalar")]
    fn f64_field_panics_on_kind_mismatch() {
        let rows = vec![(Timestamp(1), vec![Scalar::i64(7)])];
        let _ = f64_field(&rows, 0);
    }

    #[test]
    fn to_json_renders_the_canonical_form() {
        let report = RunReport {
            manifest: RunManifest {
                commit: "abc123".to_string(),
                params: vec![
                    ("sma_fast".to_string(), Scalar::i64(2)),
                    ("sma_slow".to_string(), Scalar::i64(4)),
                    ("bias_scale".to_string(), Scalar::f64(1.0)),
                ],
                defaults: vec![],
                window: (Timestamp(1), Timestamp(6)),
                seed: 0,
                broker: "sim-optimal(pip_size=1.0)".to_string(),
                selection: None,
                instrument: None,
                topology_hash: None,
                project: None,
            },
            metrics: RunMetrics {
                total_pips: 12.0,
                max_drawdown: 1.0,
                bias_sign_flips: 1,
                r: None,
            },
        };
        assert_eq!(
            report.to_json(),
            r#"{"manifest":{"commit":"abc123","params":[["sma_fast",{"I64":2}],["sma_slow",{"I64":4}],["bias_scale",{"F64":1.0}]],"defaults":[],"window":[1,6],"seed":0,"broker":"sim-optimal(pip_size=1.0)"},"metrics":{"total_pips":12.0,"max_drawdown":1.0,"bias_sign_flips":1}}"#,
        );
    }

    #[test]
    fn to_json_equals_serde_disk_shape() {
        // the same RunReport value the canonical-form test builds.
        let report = RunReport {
            manifest: RunManifest {
                commit: "abc123".to_string(),
                params: vec![
                    ("sma_fast".to_string(), Scalar::i64(2)),
                    ("sma_slow".to_string(), Scalar::i64(4)),
                    ("bias_scale".to_string(), Scalar::f64(1.0)),
                ],
                defaults: vec![],
                window: (Timestamp(1), Timestamp(6)),
                seed: 0,
                broker: "sim-optimal(pip_size=1.0)".to_string(),
                selection: None,
                instrument: None,
                topology_hash: None,
                project: None,
            },
            metrics: RunMetrics { total_pips: 12.0, max_drawdown: 1.0, bias_sign_flips: 1, r: None },
        };
        // stdout (to_json) and disk (serde_json::to_string) are now the same bytes.
        assert_eq!(report.to_json(), serde_json::to_string(&report).unwrap());
    }

    #[test]
    fn runreport_serde_round_trips() {
        let report = RunReport {
            manifest: RunManifest {
                commit: "abc123".to_string(),
                params: vec![
                    ("sma_fast".to_string(), Scalar::i64(2)),
                    ("sma_slow".to_string(), Scalar::i64(4)),
                    ("bias_scale".to_string(), Scalar::f64(1.0)),
                ],
                defaults: vec![],
                window: (Timestamp(1), Timestamp(6)),
                seed: 0,
                broker: "sim-optimal(pip_size=1.0)".to_string(),
                selection: None,
                instrument: None,
                topology_hash: None,
                project: None,
            },
            metrics: RunMetrics { total_pips: 12.0, max_drawdown: 1.0, bias_sign_flips: 1, r: None },
        };
        let json = serde_json::to_string(&report).expect("serialize RunReport");
        // window is a 2-element [from, to] array (Timestamp newtype is transparent)
        assert!(json.contains("\"window\":[1,6]"), "window shape: {json}");
        let back: RunReport<RunMetrics> = serde_json::from_str(&json).expect("deserialize RunReport");
        assert_eq!(back, report);
    }

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

    #[test]
    fn columnar_trace_round_trips_f64_rows() {
        let rows = vec![
            (Timestamp(2), vec![Scalar::f64(10.0)]),
            (Timestamp(3), vec![Scalar::f64(20.0)]),
            (Timestamp(4), vec![Scalar::f64(30.0)]),
        ];
        let ct = ColumnarTrace::from_rows("equity", &[ScalarKind::F64], &rows);
        assert_eq!(ct.tap, "equity");
        assert_eq!(ct.kinds, vec!["F64".to_string()]);
        assert_eq!(ct.ts, vec![2, 3, 4]);
        assert_eq!(ct.columns, vec![vec![10.0, 20.0, 30.0]]);
        // to_rows is the inverse for f64 taps
        assert_eq!(ct.to_rows(), rows);
    }

    #[test]
    fn columnar_trace_empty_rows_yields_empty_columns_sized_to_kinds() {
        let ct = ColumnarTrace::from_rows("exposure", &[ScalarKind::F64], &[]);
        assert_eq!(ct.ts, Vec::<i64>::new());
        assert_eq!(ct.columns, vec![Vec::<f64>::new()]);
        assert!(ct.to_rows().is_empty());
    }

    #[test]
    fn columnar_trace_coerces_non_f64_kinds_to_f64() {
        let rows = vec![
            (Timestamp(1), vec![Scalar::i64(7), Scalar::bool(true)]),
            (Timestamp(2), vec![Scalar::i64(9), Scalar::bool(false)]),
        ];
        let ct = ColumnarTrace::from_rows("mix", &[ScalarKind::I64, ScalarKind::Bool], &rows);
        assert_eq!(ct.kinds, vec!["I64".to_string(), "Bool".to_string()]);
        assert_eq!(ct.columns, vec![vec![7.0, 9.0], vec![1.0, 0.0]]);
    }

    /// The fourth coercion arm: a Timestamp-kind column tags as "Timestamp" and
    /// its epoch-ns survives the f64 projection (`scalar_to_f64` reads `as_ts().0`),
    /// closing the kind for which `from_rows` would otherwise be untested.
    #[test]
    fn columnar_trace_coerces_timestamp_kind_to_epoch_f64() {
        let rows = vec![
            (Timestamp(1), vec![Scalar::ts(Timestamp(1_700_000_000_000_000_000))]),
            (Timestamp(2), vec![Scalar::ts(Timestamp(1_700_000_000_000_000_060))]),
        ];
        let ct = ColumnarTrace::from_rows("clock", &[ScalarKind::Timestamp], &rows);
        assert_eq!(ct.kinds, vec!["Timestamp".to_string()]);
        assert_eq!(
            ct.columns,
            vec![vec![1_700_000_000_000_000_000.0, 1_700_000_000_000_000_060.0]],
        );
    }

    /// A row wider than the declared kinds is a wiring bug (a sink's column count
    /// is fixed at bootstrap), surfaced as a named panic like `f64_field` — not a
    /// bare index-out-of-bounds.
    #[test]
    #[should_panic(expected = "row width 2 disagrees with kinds.len() 1")]
    fn from_rows_panics_on_row_wider_than_kinds() {
        let rows = vec![(Timestamp(1), vec![Scalar::f64(1.0), Scalar::f64(2.0)])];
        let _ = ColumnarTrace::from_rows("wide", &[ScalarKind::F64], &rows);
    }

    /// Symmetric to the wide case: a row narrower than the declared kinds is the
    /// same wiring-bug class and carries the same named panic.
    #[test]
    #[should_panic(expected = "row width 1 disagrees with kinds.len() 2")]
    fn from_rows_panics_on_row_narrower_than_kinds() {
        let rows = vec![(Timestamp(1), vec![Scalar::f64(1.0)])];
        let _ = ColumnarTrace::from_rows("narrow", &[ScalarKind::F64, ScalarKind::F64], &rows);
    }
}