feat(trace-charts): persist recorder taps to disk (iter 1)

Iteration 1 (persist half) of the visual face — spec 0056 / plan 0055:
tap data from a graph and save it to disk so it can later be charted
(the serve/chart half is iteration 2).

- ColumnarTrace (aura-engine): a drained tap <-> columnar SoA form
  ({tap, kinds, ts, columns}); values coerced to f64 for plotting with
  the base type preserved in `kinds` (C7); to_rows rebuilds uniformly-f64
  rows for the serve-side join. Pure (C1), beside join_on_ts/summarize.
- TraceStore (aura-registry): a per-run directory under
  runs/traces/<name>/ — one <tap>.json per tap + an index.json (manifest
  + tap order), beside the families.jsonl sibling store. Missing run =
  NotFound (no panic); corrupt file = Parse{file}.
- CLI: a shared persist_traces helper threaded into run_sample /
  run_macd / run_sample_real via an orthogonal `trace: Option<&str>`;
  new arms `aura run [--macd] --trace <name>` and `--trace <name>` on
  `run --real`. The default (no --trace) run is byte-for-byte unchanged.

Verified by the orchestrator: `cargo test --workspace` green (incl. the
4 columnar_trace, 3 trace_store, 4 cli_run e2e, and the 4-tuple
parse_real_args tests); `cargo clippy --workspace --all-targets -D
warnings` exit 0. Implementer deviations folded in, all
compiler-prescribed: #[derive(Debug)] on RunTraces (the RED tests
panic-format a Result over it), #[allow(clippy::type_complexity)] on
the parse_real_args 4-tuple (mirrors the crate's sample_harness idiom).
Default-run byte-identity stays pinned by run_prints_json_and_exits_zero.

refs #101

crates/aura-engine/src/lib.rs

@@ -61,7 +61,9 @@ pub use harness::{
     window_of, BootstrapError, Edge, FlatGraph, Harness, Source, SourceSpec, SyntheticSpec, Target,
     VecSource,
 };
-pub use report::{f64_field, join_on_ts, summarize, JoinedRow, RunManifest, RunMetrics, RunReport};
+pub use report::{
+    f64_field, join_on_ts, summarize, ColumnarTrace, JoinedRow, RunManifest, RunMetrics, RunReport,
+};
 pub use sweep::{
     sweep, GridSpace, ParamRange, RandomSpace, Space, SweepError, SweepFamily, SweepPoint,
 };

crates/aura-engine/src/report.rs

@@ -186,6 +186,95 @@ pub fn join_on_ts(
         .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::*;
@@ -510,4 +599,75 @@ mod tests {
         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);
+    }
 }