From 8bb525604131ea26f764c6b72a14e82ee7ae5dc2 Mon Sep 17 00:00:00 2001 From: Brummel Date: Fri, 19 Jun 2026 00:14:18 +0200 Subject: [PATCH] feat(trace-charts): persist recorder taps to disk (iter 1) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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// — one .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 ` and `--trace ` 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-cli/src/main.rs | 184 ++++++++++++--------- crates/aura-cli/tests/cli_run.rs | 165 +++++++++++++++++++ crates/aura-engine/src/lib.rs | 4 +- crates/aura-engine/src/report.rs | 160 ++++++++++++++++++ crates/aura-registry/src/lib.rs | 3 + crates/aura-registry/src/trace_store.rs | 210 ++++++++++++++++++++++++ 6 files changed, 647 insertions(+), 79 deletions(-) create mode 100644 crates/aura-registry/src/trace_store.rs diff --git a/crates/aura-cli/src/main.rs b/crates/aura-cli/src/main.rs index a619422..5b252f3 100644 --- a/crates/aura-cli/src/main.rs +++ b/crates/aura-cli/src/main.rs @@ -15,14 +15,14 @@ mod render; use aura_core::{zip_params, Cell, Firing, Scalar, ScalarKind, Timestamp}; use aura_engine::{ - f64_field, monte_carlo, param_stability, summarize, walk_forward, window_of, Composite, Edge, - FlatGraph, GraphBuilder, Harness, McAggregate, McFamily, RollMode, RunManifest, RunReport, - SourceSpec, SweepFamily, SyntheticSpec, Target, VecSource, WalkForwardResult, WindowBounds, - WindowRoller, WindowRun, + f64_field, monte_carlo, param_stability, summarize, walk_forward, window_of, ColumnarTrace, + Composite, Edge, FlatGraph, GraphBuilder, Harness, McAggregate, McFamily, RollMode, + RunManifest, RunReport, SourceSpec, SweepFamily, SyntheticSpec, Target, VecSource, + WalkForwardResult, WindowBounds, WindowRoller, WindowRun, }; use aura_registry::{ group_families, mc_member_reports, optimize, rank_by, sweep_member_reports, - walkforward_member_reports, FamilyKind, Registry, + walkforward_member_reports, FamilyKind, Registry, TraceStore, }; use aura_std::{Ema, Exposure, LinComb, Recorder, SimBroker, Sma, Sub}; use std::sync::mpsc::{self, Receiver}; @@ -152,10 +152,31 @@ fn sim_optimal_manifest( } } +/// Persist a run's drained taps to the on-disk trace store under `runs/traces//` +/// (beside the run registry's `runs/`). Shared by every run form — all drain the same +/// two f64 taps (equity off the broker, exposure off the strategy) and carry a +/// `RunManifest`. Pure wiring over `ColumnarTrace` + `TraceStore`; the engine is +/// untouched. An I/O failure is a usage-level error (stderr + exit 2), per the spec. +fn persist_traces( + name: &str, + manifest: &RunManifest, + eq_rows: &[(Timestamp, Vec)], + ex_rows: &[(Timestamp, Vec)], +) { + let taps = vec![ + ColumnarTrace::from_rows("equity", &[ScalarKind::F64], eq_rows), + ColumnarTrace::from_rows("exposure", &[ScalarKind::F64], ex_rows), + ]; + if let Err(e) = TraceStore::open("runs").write(name, manifest, &taps) { + eprintln!("aura: trace persist failed: {e}"); + std::process::exit(2); + } +} + /// Run the sample harness and fold it into a `RunReport` (drain both sinks → /// `f64_field` → `summarize` → pair with a `RunManifest`). Pure and deterministic /// (C1): the same build yields the same report. -fn run_sample() -> RunReport { +fn run_sample(trace: Option<&str>) -> RunReport { let (mut h, rx_eq, rx_ex) = sample_harness(SYNTHETIC_PIP_SIZE); let sources: Vec> = vec![Box::new(VecSource::new(synthetic_prices()))]; @@ -164,23 +185,21 @@ fn run_sample() -> RunReport { let eq_rows: Vec<(Timestamp, Vec)> = rx_eq.try_iter().collect(); let ex_rows: Vec<(Timestamp, Vec)> = rx_ex.try_iter().collect(); - let equity = f64_field(&eq_rows, 0); - let exposure = f64_field(&ex_rows, 0); - let metrics = summarize(&equity, &exposure); - - RunReport { - manifest: sim_optimal_manifest( - vec![ - ("sma_fast".to_string(), Scalar::i64(2)), - ("sma_slow".to_string(), Scalar::i64(4)), - ("exposure_scale".to_string(), Scalar::f64(0.5)), - ], - window, - 0, - SYNTHETIC_PIP_SIZE, - ), - metrics, + let manifest = sim_optimal_manifest( + vec![ + ("sma_fast".to_string(), Scalar::i64(2)), + ("sma_slow".to_string(), Scalar::i64(4)), + ("exposure_scale".to_string(), Scalar::f64(0.5)), + ], + window, + 0, + SYNTHETIC_PIP_SIZE, + ); + if let Some(name) = trace { + persist_traces(name, &manifest, &eq_rows, &ex_rows); } + let metrics = summarize(&f64_field(&eq_rows, 0), &f64_field(&ex_rows, 0)); + RunReport { manifest, metrics } } /// `aura run --real `: run the built-in sample harness over real M1 close @@ -190,7 +209,7 @@ fn run_sample() -> RunReport { /// (a Source is single-pass), so the run source streams the window untouched. /// A no-local-data condition (unknown symbol, or a window overlapping no file / no /// bars) is a usage error: stderr + exit(2), not a panic. -fn run_sample_real(symbol: &str, from_ms: Option, to_ms: Option) -> RunReport { +fn run_sample_real(symbol: &str, from_ms: Option, to_ms: Option, trace: Option<&str>) -> RunReport { // Per-instrument pip: look up BEFORE any data access, so an un-specced symbol // refuses without touching local data. Honest by construction. let spec = match aura_ingest::instrument_spec(symbol) { @@ -237,23 +256,21 @@ fn run_sample_real(symbol: &str, from_ms: Option, to_ms: Option) -> Ru let eq_rows: Vec<(Timestamp, Vec)> = rx_eq.try_iter().collect(); let ex_rows: Vec<(Timestamp, Vec)> = rx_ex.try_iter().collect(); - let equity = f64_field(&eq_rows, 0); - let exposure = f64_field(&ex_rows, 0); - let metrics = summarize(&equity, &exposure); - - RunReport { - manifest: sim_optimal_manifest( - vec![ - ("sma_fast".to_string(), Scalar::i64(2)), - ("sma_slow".to_string(), Scalar::i64(4)), - ("exposure_scale".to_string(), Scalar::f64(0.5)), - ], - window, - 0, - spec.pip_size, - ), - metrics, + let manifest = sim_optimal_manifest( + vec![ + ("sma_fast".to_string(), Scalar::i64(2)), + ("sma_slow".to_string(), Scalar::i64(4)), + ("exposure_scale".to_string(), Scalar::f64(0.5)), + ], + window, + 0, + spec.pip_size, + ); + if let Some(name) = trace { + persist_traces(name, &manifest, &eq_rows, &ex_rows); } + let metrics = summarize(&f64_field(&eq_rows, 0), &f64_field(&ex_rows, 0)); + RunReport { manifest, metrics } } /// Parse the `run --real` tail: a mandatory `` then zero-or-more @@ -261,25 +278,29 @@ fn run_sample_real(symbol: &str, from_ms: Option, to_ms: Option) -> Ru /// arg grammar is unit-testable; `main` does the side effects. An unknown token, a /// flag without its value, a non-`i64` ms, or a repeated flag is an `Err` carrying /// a short usage message. -fn parse_real_args(rest: &[&str]) -> Result<(String, Option, Option), String> { - let usage = || "run --real [--from ] [--to ]".to_string(); +#[allow(clippy::type_complexity)] +fn parse_real_args( + rest: &[&str], +) -> Result<(String, Option, Option, Option), String> { + let usage = || "run --real [--from ] [--to ] [--trace ]".to_string(); let (symbol, mut tail) = match rest.split_first() { Some((sym, t)) if !sym.is_empty() => (*sym, t), _ => return Err(usage()), }; let mut from: Option = None; let mut to: Option = None; + let mut trace: Option = None; while let Some((flag, t)) = tail.split_first() { let (value, t) = t.split_first().ok_or_else(usage)?; - let ms: i64 = value.parse().map_err(|_| usage())?; match *flag { - "--from" if from.is_none() => from = Some(ms), - "--to" if to.is_none() => to = Some(ms), + "--from" if from.is_none() => from = Some(value.parse().map_err(|_| usage())?), + "--to" if to.is_none() => to = Some(value.parse().map_err(|_| usage())?), + "--trace" if trace.is_none() => trace = Some((*value).to_string()), _ => return Err(usage()), } tail = t; } - Ok((symbol.to_string(), from, to)) + Ok((symbol.to_string(), from, to, trace)) } /// The SMA-cross signal as a named composite (price -> fast/slow SMA -> spread). @@ -820,7 +841,7 @@ fn macd_prices() -> Vec<(Timestamp, Scalar)> { /// (the same bootstrap path the SMA sample's compiled view uses), drive it on the /// synthetic stream, and fold both sinks into a `RunReport`. Pure and /// deterministic (C1). -fn run_macd() -> RunReport { +fn run_macd(trace: Option<&str>) -> RunReport { let (tx_eq, rx_eq) = mpsc::channel(); let (tx_ex, rx_ex) = mpsc::channel(); let flat = macd_strategy_blueprint(tx_eq, tx_ex) @@ -835,28 +856,26 @@ fn run_macd() -> RunReport { let eq_rows: Vec<(Timestamp, Vec)> = rx_eq.try_iter().collect(); let ex_rows: Vec<(Timestamp, Vec)> = rx_ex.try_iter().collect(); - let equity = f64_field(&eq_rows, 0); - let exposure = f64_field(&ex_rows, 0); - let metrics = summarize(&equity, &exposure); - - RunReport { - manifest: sim_optimal_manifest( - vec![ - ("ema_fast".to_string(), Scalar::i64(2)), - ("ema_slow".to_string(), Scalar::i64(4)), - ("ema_signal".to_string(), Scalar::i64(3)), - ("exposure_scale".to_string(), Scalar::f64(0.5)), - ], - window, - 0, - SYNTHETIC_PIP_SIZE, - ), - metrics, + let manifest = sim_optimal_manifest( + vec![ + ("ema_fast".to_string(), Scalar::i64(2)), + ("ema_slow".to_string(), Scalar::i64(4)), + ("ema_signal".to_string(), Scalar::i64(3)), + ("exposure_scale".to_string(), Scalar::f64(0.5)), + ], + window, + 0, + SYNTHETIC_PIP_SIZE, + ); + if let Some(name) = trace { + persist_traces(name, &manifest, &eq_rows, &ex_rows); } + let metrics = summarize(&f64_field(&eq_rows, 0), &f64_field(&ex_rows, 0)); + RunReport { manifest, metrics } } const USAGE: &str = - "usage: aura run [--macd] | aura run --real [--from ] [--to ] | aura graph | aura sweep [--name ] | aura mc [--name ] | aura walkforward [--name ] | aura runs families | aura runs family [rank ]"; + "usage: aura run [--macd] [--trace ] | aura run --real [--from ] [--to ] [--trace ] | aura graph | aura sweep [--name ] | aura mc [--name ] | aura walkforward [--name ] | aura runs families | aura runs family [rank ]"; fn main() { // Collect argv and match the whole vector: every accepted form is exhaustive, @@ -864,10 +883,14 @@ fn main() { // than masquerading as a successful run (#16 strict reading). let args: Vec = std::env::args().skip(1).collect(); match args.iter().map(String::as_str).collect::>().as_slice() { - ["run"] => println!("{}", run_sample().to_json()), - ["run", "--macd"] => println!("{}", run_macd().to_json()), + ["run"] => println!("{}", run_sample(None).to_json()), + ["run", "--macd"] => println!("{}", run_macd(None).to_json()), + ["run", "--trace", name] => println!("{}", run_sample(Some(name)).to_json()), + ["run", "--macd", "--trace", name] => println!("{}", run_macd(Some(name)).to_json()), ["run", "--real", rest @ ..] => match parse_real_args(rest) { - Ok((sym, from, to)) => println!("{}", run_sample_real(&sym, from, to).to_json()), + Ok((sym, from, to, trace)) => { + println!("{}", run_sample_real(&sym, from, to, trace.as_deref()).to_json()) + } Err(msg) => { eprintln!("aura: {msg}"); std::process::exit(2); @@ -1096,8 +1119,8 @@ mod tests { // the MACD strategy authors a nested EMA-of-EMA composite, compiles it to a // flat runnable harness (the call not panicking proves the compile+bootstrap // path), and runs it. C1 determinism: two runs are bit-identical. - let r1 = run_macd(); - let r2 = run_macd(); + let r1 = run_macd(None); + let r2 = run_macd(None); assert_eq!(r1.metrics, r2.metrics); assert_eq!(r1.to_json(), r2.to_json()); @@ -1177,8 +1200,8 @@ mod tests { // The headline: a real-data run over the bounded Sept-2024 window yields // a finite, C1-deterministic RunReport. - let r1 = run_sample_real(SYMBOL, Some(GER40_SEP2024_FROM_MS), Some(GER40_SEP2024_TO_MS)); - let r2 = run_sample_real(SYMBOL, Some(GER40_SEP2024_FROM_MS), Some(GER40_SEP2024_TO_MS)); + let r1 = run_sample_real(SYMBOL, Some(GER40_SEP2024_FROM_MS), Some(GER40_SEP2024_TO_MS), None); + let r2 = run_sample_real(SYMBOL, Some(GER40_SEP2024_FROM_MS), Some(GER40_SEP2024_TO_MS), None); assert!( r1.metrics.total_pips.is_finite(), @@ -1212,12 +1235,12 @@ mod tests { // Bounded to the vetted Sept-2024 window so the pip-label + determinism // checks run fast (not the full unbounded archive). let report = - run_sample_real("GER40", Some(GER40_SEP2024_FROM_MS), Some(GER40_SEP2024_TO_MS)); + run_sample_real("GER40", Some(GER40_SEP2024_FROM_MS), Some(GER40_SEP2024_TO_MS), None); // The looked-up index pip (1.0) reaches the manifest — not the FX 0.0001. assert_eq!(report.manifest.broker, "sim-optimal(pip_size=1)"); // Deterministic (C1): a second run yields the same report. let again = - run_sample_real("GER40", Some(GER40_SEP2024_FROM_MS), Some(GER40_SEP2024_TO_MS)); + run_sample_real("GER40", Some(GER40_SEP2024_FROM_MS), Some(GER40_SEP2024_TO_MS), None); assert_eq!(report.manifest.broker, again.manifest.broker); assert_eq!(report.metrics.total_pips, again.metrics.total_pips); } @@ -1227,15 +1250,20 @@ mod tests { /// value — the pure arg grammar `main` relies on for `run --real`. #[test] fn parse_real_args_accepts_symbol_and_optional_window() { - assert_eq!(parse_real_args(&["EURUSD"]), Ok(("EURUSD".to_string(), None, None))); + assert_eq!(parse_real_args(&["EURUSD"]), Ok(("EURUSD".to_string(), None, None, None))); assert_eq!( parse_real_args(&["EURUSD", "--from", "100", "--to", "200"]), - Ok(("EURUSD".to_string(), Some(100), Some(200))) + Ok(("EURUSD".to_string(), Some(100), Some(200), None)) ); // flags in any order assert_eq!( parse_real_args(&["EURUSD", "--to", "200", "--from", "100"]), - Ok(("EURUSD".to_string(), Some(100), Some(200))) + Ok(("EURUSD".to_string(), Some(100), Some(200), None)) + ); + // the --trace tail flag carries a string name, parsed alongside the window + assert_eq!( + parse_real_args(&["EURUSD", "--from", "100", "--trace", "demo"]), + Ok(("EURUSD".to_string(), Some(100), None, Some("demo".to_string()))) ); // a flag without its value, an empty symbol, and a non-numeric ms all reject assert!(parse_real_args(&["EURUSD", "--from"]).is_err()); @@ -1245,8 +1273,8 @@ mod tests { #[test] fn run_sample_is_deterministic_and_non_trivial() { - let r1 = run_sample(); - let r2 = run_sample(); + let r1 = run_sample(None); + let r2 = run_sample(None); // C1 determinism: two runs are bit-identical (metrics + rendered JSON). assert_eq!(r1.metrics, r2.metrics); assert_eq!(r1.to_json(), r2.to_json()); diff --git a/crates/aura-cli/tests/cli_run.rs b/crates/aura-cli/tests/cli_run.rs index 6cb7d62..acb7edc 100644 --- a/crates/aura-cli/tests/cli_run.rs +++ b/crates/aura-cli/tests/cli_run.rs @@ -248,6 +248,171 @@ fn run_real_vetted_index_pip_reaches_the_emitted_manifest() { ); } +#[test] +fn run_trace_persists_taps_and_plain_run_writes_no_traces() { + let cwd = temp_cwd("run-trace"); + + // plain `aura run` (no --trace) persists nothing to disk. + let plain = Command::new(BIN).arg("run").current_dir(&cwd).output().expect("spawn aura run"); + assert!(plain.status.success(), "plain run exit: {:?}", plain.status); + assert!(!cwd.join("runs/traces").exists(), "plain run must write no trace files"); + + // `aura run --trace demo` persists the two taps + index, and still prints the report. + let traced = Command::new(BIN) + .args(["run", "--trace", "demo"]) + .current_dir(&cwd) + .output() + .expect("spawn aura run --trace"); + assert!(traced.status.success(), "traced run exit: {:?}", traced.status); + let stdout = String::from_utf8(traced.stdout).expect("utf-8 stdout"); + assert!(stdout.trim_end().starts_with("{\"manifest\":{"), "report still printed: {stdout:?}"); + assert!(cwd.join("runs/traces/demo/index.json").exists(), "index.json written"); + assert!(cwd.join("runs/traces/demo/equity.json").exists(), "equity tap written"); + assert!(cwd.join("runs/traces/demo/exposure.json").exists(), "exposure tap written"); + + let _ = std::fs::remove_dir_all(&cwd); +} + +/// Property: a persisted `.json` is the **columnar SoA form** a downstream +/// chart can consume, not an opaque blob — it carries the four +/// `tap`/`kinds`/`ts`/`columns` keys, the equity tap is kind-tagged `["F64"]` +/// (C7: the base type survives on disk), and the SoA invariant holds: `ts` has one +/// entry per recorded cycle and EVERY column has exactly that many entries (parallel +/// arrays, one timestamp per row). This is acceptance criterion 2 at the built-binary +/// file boundary — the unit round-trip pins the in-memory transpose, but only an +/// end-to-end run pins that the bytes actually written to disk hold that shape. +/// Asserts on structure + array-length parity, never on the exact float values +/// (pip equity's low bits vary build-to-build), so it stays deterministic. +#[test] +fn run_trace_persists_columnar_soa_shape_on_disk() { + let cwd = temp_cwd("run-trace-shape"); + let out = Command::new(BIN) + .args(["run", "--trace", "demo"]) + .current_dir(&cwd) + .output() + .expect("spawn aura run --trace"); + assert!(out.status.success(), "traced run exit: {:?}", out.status); + + let equity = std::fs::read_to_string(cwd.join("runs/traces/demo/equity.json")) + .expect("read equity.json"); + // The four SoA keys are present, equity is f64-kind-tagged, and the parallel + // arrays open as expected (the synthetic sample harness has a single column). + assert!(equity.contains("\"tap\":\"equity\""), "tap name missing: {equity}"); + assert!(equity.contains("\"kinds\":[\"F64\"]"), "F64 kind tag missing: {equity}"); + assert!(equity.contains("\"ts\":["), "ts axis missing: {equity}"); + assert!(equity.contains("\"columns\":[["), "columns array missing: {equity}"); + + // SoA invariant: one timestamp per row, and every column has exactly that many + // entries. Count the ts entries (the synthetic window is 7 cycles) and the + // single column's entries by their comma-separated cardinality. + let ts_body = json_array_body(&equity, "\"ts\":["); + let col_body = json_array_body(&equity, "\"columns\":[["); + let ts_len = ts_body.split(',').count(); + let col_len = col_body.split(',').count(); + assert_eq!(ts_len, 7, "synthetic run records 7 cycles; ts was: {ts_body:?}"); + assert_eq!( + col_len, ts_len, + "SoA parity broken: ts has {ts_len} entries but the column has {col_len}; file: {equity}" + ); + + let _ = std::fs::remove_dir_all(&cwd); +} + +/// Property: `index.json` is the run's **table of contents** — it carries the run's +/// `RunManifest` (so a chart page can show run context: commit, broker, window) and +/// the tap names in a **deterministic order** (`["equity","exposure"]`). The serve +/// half (`aura chart`, iteration 2) reads taps in this index order; a regression that +/// dropped the manifest, or made the tap list order-unstable, would silently corrupt +/// the chart header / the per-series mapping. Asserts on the manifest's structural +/// presence (the commit value is the volatile git HEAD, pinned elsewhere) and the +/// exact deterministic tap order. +#[test] +fn run_trace_index_carries_manifest_and_ordered_tap_list() { + let cwd = temp_cwd("run-trace-index"); + let out = Command::new(BIN) + .args(["run", "--trace", "demo"]) + .current_dir(&cwd) + .output() + .expect("spawn aura run --trace"); + assert!(out.status.success(), "traced run exit: {:?}", out.status); + + let index = std::fs::read_to_string(cwd.join("runs/traces/demo/index.json")) + .expect("read index.json"); + // The manifest is embedded with its identifying context for the chart header. + assert!(index.contains("\"manifest\":{\"commit\":\""), "manifest missing: {index}"); + assert!(index.contains("\"window\":[1,7]"), "window missing from manifest: {index}"); + assert!( + index.contains("\"broker\":\"sim-optimal(pip_size=0.0001)\""), + "broker label missing: {index}" + ); + // The tap list is the deterministic read order the serve path depends on. + assert!( + index.contains("\"taps\":[\"equity\",\"exposure\"]"), + "tap order must be deterministic [equity, exposure]: {index}" + ); + + let _ = std::fs::remove_dir_all(&cwd); +} + +/// Property: `--trace` is an **orthogonal modifier** — it composes with `aura run +/// --real `, not only plain `run`, and is strictly **additive**: the looked-up +/// instrument pip still reaches stdout byte-for-byte while the same two taps land on +/// disk. The spec's headline is "one shared persist helper, every run form"; the +/// existing E2E only exercised the synthetic `run` path. Gated on local GER40 data +/// (skip cleanly when absent, the project convention) so it never fails on a +/// data-less machine; the no-data path is the distinct "no local data" refusal. +#[test] +fn run_real_with_trace_persists_taps_and_keeps_stdout_unchanged() { + const FROM_MS: &str = "1725148800000"; + const TO_MS: &str = "1727740799999"; + let cwd = temp_cwd("run-real-trace"); + let out = Command::new(BIN) + .args(["run", "--real", "GER40", "--from", FROM_MS, "--to", TO_MS, "--trace", "ger"]) + .current_dir(&cwd) + .output() + .expect("spawn aura run --real --trace"); + + // Skip on a data-less machine: vetted GER40 with no local data takes the + // distinct no-data path (exit 2), never the pip-refusal, and writes no traces. + if out.status.code() == Some(2) { + let stderr = String::from_utf8_lossy(&out.stderr); + assert!( + stderr.contains("no local data for symbol 'GER40'"), + "exit 2 for vetted GER40 --trace must be the no-data path, got: {stderr}" + ); + eprintln!("skip: no local GER40 data"); + let _ = std::fs::remove_dir_all(&cwd); + return; + } + + assert_eq!(out.status.code(), Some(0), "real --trace exit: {:?}", out.status); + // Additive: the index-pip report still reaches stdout unchanged. + let stdout = String::from_utf8(out.stdout).expect("utf-8 stdout"); + let line = stdout.trim_end(); + assert!(line.starts_with("{\"manifest\":{"), "report still printed: {line}"); + assert!( + line.contains("\"broker\":\"sim-optimal(pip_size=1)\""), + "real --trace must keep the GER40 index pip on stdout: {line}" + ); + // And the same two taps landed under the named run dir. + assert!(cwd.join("runs/traces/ger/index.json").exists(), "index.json written"); + assert!(cwd.join("runs/traces/ger/equity.json").exists(), "equity tap written"); + assert!(cwd.join("runs/traces/ger/exposure.json").exists(), "exposure tap written"); + + let _ = std::fs::remove_dir_all(&cwd); +} + +/// Extract the body of a JSON array opened by `marker` (e.g. `"ts":[`) up to its +/// closing `]`, by substring — no serde dependency, mirroring the sibling tests' +/// substring parsing. Used only to count comma-separated entries for the SoA +/// length-parity check. +fn json_array_body<'a>(json: &'a str, marker: &str) -> &'a str { + let start = json.find(marker).expect("array marker present") + marker.len(); + let rest = &json[start..]; + let end = rest.find(']').expect("array is closed"); + &rest[..end] +} + #[test] fn no_args_prints_usage_and_exits_two() { let out = Command::new(BIN).output().expect("spawn aura"); diff --git a/crates/aura-engine/src/lib.rs b/crates/aura-engine/src/lib.rs index 863a853..479189c 100644 --- a/crates/aura-engine/src/lib.rs +++ b/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, }; diff --git a/crates/aura-engine/src/report.rs b/crates/aura-engine/src/report.rs index afd73c3..22df460 100644 --- a/crates/aura-engine/src/report.rs +++ b/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, + pub ts: Vec, + pub columns: Vec>, +} + +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)]) -> Self { + let ts: Vec = rows.iter().map(|(t, _)| t.0).collect(); + let mut columns: Vec> = 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)> { + 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::::new()); + assert_eq!(ct.columns, vec![Vec::::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); + } } diff --git a/crates/aura-registry/src/lib.rs b/crates/aura-registry/src/lib.rs index b8ad8b2..4f7d1ae 100644 --- a/crates/aura-registry/src/lib.rs +++ b/crates/aura-registry/src/lib.rs @@ -27,6 +27,9 @@ pub use lineage::{ FamilyKind, FamilyRunRecord, }; +mod trace_store; +pub use trace_store::{RunTraces, TraceStore, TraceStoreError}; + /// An append-only run registry over a JSONL file: one serde_json line per /// `RunReport`. pub struct Registry { diff --git a/crates/aura-registry/src/trace_store.rs b/crates/aura-registry/src/trace_store.rs new file mode 100644 index 0000000..aa4acea --- /dev/null +++ b/crates/aura-registry/src/trace_store.rs @@ -0,0 +1,210 @@ +//! The per-run trace store: a directory of columnar tap files under +//! `/traces//`, beside the run registry's `runs.jsonl`. One +//! subdirectory per run name; one `.json` (a `ColumnarTrace`) per tap; plus +//! `index.json` carrying the run's `RunManifest` and the tap order. Mirrors the +//! registry's directory-co-located sibling-store discipline; a missing run reads as +//! `NotFound` (treat-as-absent), never a panic. + +use std::fmt; +use std::fs; +use std::path::{Path, PathBuf}; + +use aura_engine::{ColumnarTrace, RunManifest}; +use serde::{Deserialize, Serialize}; + +/// A per-run trace directory rooted at `/traces`. One subdir per run name. +pub struct TraceStore { + dir: PathBuf, +} + +/// A run's read-back traces: its manifest plus its taps in index (write) order. +#[derive(Debug)] +pub struct RunTraces { + pub manifest: RunManifest, + pub taps: Vec, +} + +/// `index.json`: the run manifest + the tap names in write order (= read order). +#[derive(Serialize, Deserialize)] +struct Index { + manifest: RunManifest, + taps: Vec, +} + +impl TraceStore { + /// Bind to `/traces`. No I/O — directories are created on write. + pub fn open(runs_dir: impl AsRef) -> TraceStore { + TraceStore { dir: runs_dir.as_ref().join("traces") } + } + + /// Write one run's taps + index under `traces//`, creating directories as + /// needed. Each tap lands in `.json`; the manifest + tap order in + /// `index.json`. + pub fn write( + &self, + name: &str, + manifest: &RunManifest, + taps: &[ColumnarTrace], + ) -> Result<(), TraceStoreError> { + let run_dir = self.dir.join(name); + fs::create_dir_all(&run_dir)?; + for tap in taps { + let path = run_dir.join(format!("{}.json", tap.tap)); + let body = serde_json::to_string(tap).map_err(|source| TraceStoreError::Parse { + file: path.display().to_string(), + source, + })?; + fs::write(&path, body)?; + } + let index = Index { + manifest: manifest.clone(), + taps: taps.iter().map(|t| t.tap.clone()).collect(), + }; + let index_path = run_dir.join("index.json"); + let body = serde_json::to_string(&index).map_err(|source| TraceStoreError::Parse { + file: index_path.display().to_string(), + source, + })?; + fs::write(&index_path, body)?; + Ok(()) + } + + /// Read a run's index + taps back, in index order. A missing run directory (no + /// `index.json`) is `NotFound`, not a panic; a malformed file is `Parse{file,..}`. + pub fn read(&self, name: &str) -> Result { + let run_dir = self.dir.join(name); + let index_path = run_dir.join("index.json"); + let index_text = match fs::read_to_string(&index_path) { + Ok(t) => t, + Err(e) if e.kind() == std::io::ErrorKind::NotFound => { + return Err(TraceStoreError::NotFound(name.to_string())) + } + Err(e) => return Err(TraceStoreError::Io(e)), + }; + let index: Index = serde_json::from_str(&index_text).map_err(|source| { + TraceStoreError::Parse { file: index_path.display().to_string(), source } + })?; + let mut taps = Vec::with_capacity(index.taps.len()); + for tap_name in &index.taps { + let path = run_dir.join(format!("{tap_name}.json")); + let text = fs::read_to_string(&path)?; + let tap: ColumnarTrace = serde_json::from_str(&text).map_err(|source| { + TraceStoreError::Parse { file: path.display().to_string(), source } + })?; + taps.push(tap); + } + Ok(RunTraces { manifest: index.manifest, taps }) + } +} + +/// What can go wrong reading or writing the trace store. +#[derive(Debug)] +pub enum TraceStoreError { + /// An I/O error reading or writing a trace file. + Io(std::io::Error), + /// A trace/index file did not parse (the offending file named). + Parse { file: String, source: serde_json::Error }, + /// No recorded run of this name (no `index.json` under `traces//`). + NotFound(String), +} + +impl fmt::Display for TraceStoreError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + TraceStoreError::Io(e) => write!(f, "trace store i/o: {e}"), + TraceStoreError::Parse { file, source } => { + write!(f, "trace store parse error in {file}: {source}") + } + TraceStoreError::NotFound(name) => { + write!(f, "no recorded run '{name}' under runs/traces") + } + } + } +} + +impl std::error::Error for TraceStoreError {} + +impl From for TraceStoreError { + fn from(e: std::io::Error) -> Self { + TraceStoreError::Io(e) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use aura_core::{Scalar, ScalarKind, Timestamp}; + + fn temp_traces_root(name: &str) -> PathBuf { + let dir = + std::env::temp_dir().join(format!("aura-tracestore-{}-{}", std::process::id(), name)); + let _ = fs::remove_dir_all(&dir); + fs::create_dir_all(&dir).expect("create temp traces root"); + dir + } + + fn sample_manifest() -> RunManifest { + RunManifest { + commit: "c".to_string(), + params: vec![("p".to_string(), Scalar::f64(1.0))], + window: (Timestamp(1), Timestamp(5)), + seed: 0, + broker: "sim-optimal(pip_size=0.0001)".to_string(), + } + } + + fn sample_taps() -> Vec { + vec![ + ColumnarTrace::from_rows( + "equity", + &[ScalarKind::F64], + &[(Timestamp(1), vec![Scalar::f64(0.0)]), (Timestamp(2), vec![Scalar::f64(0.5)])], + ), + ColumnarTrace::from_rows( + "exposure", + &[ScalarKind::F64], + &[(Timestamp(1), vec![Scalar::f64(1.0)]), (Timestamp(2), vec![Scalar::f64(-1.0)])], + ), + ] + } + + #[test] + fn write_then_read_round_trips() { + let root = temp_traces_root("roundtrip"); + let store = TraceStore::open(&root); + let manifest = sample_manifest(); + let taps = sample_taps(); + store.write("demo", &manifest, &taps).expect("write"); + let back = store.read("demo").expect("read"); + assert_eq!(back.manifest, manifest); + assert_eq!(back.taps, taps); + assert!(root.join("traces/demo/index.json").exists()); + assert!(root.join("traces/demo/equity.json").exists()); + assert!(root.join("traces/demo/exposure.json").exists()); + let _ = fs::remove_dir_all(&root); + } + + #[test] + fn read_missing_run_is_not_found() { + let root = temp_traces_root("missing"); + let store = TraceStore::open(&root); + match store.read("nope") { + Err(TraceStoreError::NotFound(name)) => assert_eq!(name, "nope"), + other => panic!("expected NotFound, got {other:?}"), + } + let _ = fs::remove_dir_all(&root); + } + + #[test] + fn corrupt_tap_file_is_a_parse_error_naming_the_file() { + let root = temp_traces_root("corrupt"); + let store = TraceStore::open(&root); + store.write("demo", &sample_manifest(), &sample_taps()).expect("write"); + fs::write(root.join("traces/demo/equity.json"), "not json").expect("clobber"); + match store.read("demo") { + Err(TraceStoreError::Parse { file, .. }) => assert!(file.ends_with("equity.json")), + other => panic!("expected Parse, got {other:?}"), + } + let _ = fs::remove_dir_all(&root); + } +}