From 46a85d3ea405980722099d1d78a69bf04197d425 Mon Sep 17 00:00:00 2001 From: Brummel Date: Sat, 4 Jul 2026 04:16:22 +0200 Subject: [PATCH] =?UTF-8?q?feat(cli):=20persist=5Fcampaign=5Ftraces=20?= =?UTF-8?q?=E2=80=94=20campaign=20presentation=20persists=20nominee=20trac?= =?UTF-8?q?es=20(0109=20tasks=203-4)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The deferral line dies; persist_taps is honored: after execute() returns with a claimed trace_name, the CLI re-runs each cell's nominee once in non-reduce mode (all four channels drained; windowed to the nominee manifest's own ns bounds; bound via the shipped point_from_params inverse against the wrapped probe space), asserts METRICS equality against the recorded nominee (the C1 drift alarm — hard refusal on divergence), and writes the requested-and-producible taps through the existing TraceStore as traces/{campaign8}-{run}/{strategy8}-{instrument}-w{n}/{tap}.json (ensure_name_free once; window ordinal doc-derived by position match). Loud lines: per-cell no-nominee skip, per-run net_r_equity unproducible skip (no cost leg in the campaign runner), one summary. The Task-1(b) seam test tightened to pin the deferral line's absence. Gated e2e extended: persist_taps [equity, r_equity] over GER40 — trace_name in the record, both tap files on disk, aura chart reads the cell back — ran its FULL path on this host (local archive present), not the skip branch. Gates: workspace 1041/0, clippy -D warnings clean. closes #201 --- crates/aura-cli/src/campaign_run.rs | 276 +++++++++++++++++++++++-- crates/aura-cli/tests/research_docs.rs | 144 ++++++++++++- 2 files changed, 401 insertions(+), 19 deletions(-) diff --git a/crates/aura-cli/src/campaign_run.rs b/crates/aura-cli/src/campaign_run.rs index f83451e..7853117 100644 --- a/crates/aura-cli/src/campaign_run.rs +++ b/crates/aura-cli/src/campaign_run.rs @@ -15,16 +15,19 @@ //! visible to child modules without promotion. use std::path::{Path, PathBuf}; -use std::sync::Arc; +use std::sync::{mpsc, Arc}; -use aura_campaign::{CellSpec, ExecFault, MemberFault, MemberRunner}; -use aura_core::{Cell, ParamSpec, Scalar}; -use aura_engine::{blueprint_from_json, FamilySelection, RunReport}; +use aura_campaign::{CampaignOutcome, CellSpec, ExecFault, MemberFault, MemberRunner}; +use aura_core::{Cell, ParamSpec, Scalar, ScalarKind, Timestamp}; +use aura_engine::{ + blueprint_from_json, f64_field, summarize, summarize_r, ColumnarTrace, FamilySelection, + RunReport, +}; use aura_ingest::{instrument_geometry, unix_ms_to_epoch_ns, M1Field, M1FieldSource}; -use aura_registry::CampaignRunRecord; +use aura_registry::{CampaignRunRecord, WriteKind}; use aura_research::{ campaign_to_json, content_id_of, parse_campaign, parse_process, validate_campaign, - validate_process, DocRef, + validate_process, CampaignDoc, DocRef, }; use crate::project::Env; @@ -357,16 +360,6 @@ pub(crate) fn run_campaign(target: &str, env: &Env) -> Result<(), String> { strategies.push((content_id_of(&canonical), canonical)); } - // Loud deferral (#198 decision 6), once per run — pinned ORDER: this - // prints after the document gates and BEFORE any member executes, so a - // data refusal inside execute cannot swallow it. - if !campaign.presentation.persist_taps.is_empty() { - eprintln!( - "aura: persist_taps not yet honored ({} tap(s) ignored)", - campaign.presentation.persist_taps.len() - ); - } - let runner = CliMemberRunner { env, server: Arc::new(data_server::DataServer::new(env.data_path())), @@ -445,6 +438,228 @@ pub(crate) fn run_campaign(target: &str, env: &Env) -> Result<(), String> { serde_json::to_string(&CampaignRunLine { campaign_run: &outcome.record }) .expect("campaign run record serializes") ); + + // Trace persistence runs LAST, after the always-on record line (0109): + // stdout stays data-pure (the record line is the wire contract) and the + // trace surface is a stderr concern; the realization record is already + // stored by `execute`, so a trace failure exits 1 without un-recording it. + if let Some(trace_name) = &outcome.record.trace_name { + persist_campaign_traces( + trace_name, + &campaign.presentation.persist_taps, + &outcome, + &campaign, + &strategies, + &runner.server, + env, + )?; + } + Ok(()) +} + +/// Which drained wrap-convention channel a requested tap persists from on a +/// campaign trace re-run. The routing mirrors `persist_traces_r` (main.rs): +/// equity <- the SimBroker equity recorder (tx_eq), exposure <- the Bias +/// recorder (tx_ex), r_equity <- the cum_realized_r + unrealized_r recorder +/// (tx_req). `net_r_equity` has no channel here: the campaign runner wires +/// no cost leg (`wrap_r(.., cost: None)`), so a net curve is not produced +/// by this run's configuration. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum TapChannel { + Equity, + Exposure, + REquity, +} + +/// Route one requested tap of the closed vocabulary to its drained channel; +/// `None` marks a tap this run cannot produce (the caller skips it loudly). +/// An unknown name cannot reach here (`validate_campaign` refuses it) and +/// maps to `None` all the same rather than guessing. +fn tap_channel(tap: &str) -> Option { + match tap { + "equity" => Some(TapChannel::Equity), + "exposure" => Some(TapChannel::Exposure), + "r_equity" => Some(TapChannel::REquity), + _ => None, + } +} + +/// The content-derived on-disk key of one campaign cell under +/// `traces//` — the `member_key` discipline (content, never a +/// runtime ordinal): strategy content-id prefix + instrument + +/// doc-positional window ordinal, sanitized to one portable path component. +fn campaign_cell_key(strategy: &str, instrument: &str, window_ordinal: usize) -> String { + let strategy8 = strategy.get(..8).unwrap_or(strategy); + crate::sanitize_component(&format!("{strategy8}-{instrument}-w{window_ordinal}")) +} + +/// Persist the requested `persist_taps` for every nominee cell under +/// `traces///` (0109, #201 d5): re-run each nominee +/// once in non-reduce trace mode over its own recorded `manifest.window`, +/// assert the re-run METRICS equal the recorded nominee metrics (the C1 +/// drift alarm — manifest fields are fresh-context and not compared), and +/// write the requested-AND-producible taps through the sweep verbs' +/// member-dir mechanism: a slash-joined `"{name}/{key}"` handed to +/// `TraceStore::write` (the `persist_traces_r(&format!("{name}/{key}"), ..)` +/// layout — the fn itself is not reused: it writes a fixed tap set and +/// process-exits on error). The written index manifest is the RECORDED +/// nominee's — the trace documents that run's provenance, not a re-derived +/// fresh-context one. Loud stderr per no-nominee cell and ONCE per +/// unproducible requested tap (producibility is run-configuration-level, +/// not per-cell); one summary line at the end. Every `Err` is a refusal +/// `campaign_cmd` renders as `aura: {msg}` + exit 1. +fn persist_campaign_traces( + trace_name: &str, + taps: &[String], + outcome: &CampaignOutcome, + campaign: &CampaignDoc, + strategies: &[(String, String)], + server: &Arc, + env: &Env, +) -> Result<(), String> { + let store = env.trace_store(); + store + .ensure_name_free(trace_name, WriteKind::Family) + .map_err(|e| e.to_string())?; + + // Requested ∩ producible, in request order. When nothing producible was + // requested (net_r_equity only), no member dir is written and the + // summary honestly reports 0 tap(s). + let mut routed: Vec<(&str, TapChannel)> = Vec::new(); + for tap in taps { + match tap_channel(tap) { + Some(ch) => routed.push((tap.as_str(), ch)), + None => eprintln!( + "aura: tap \"{tap}\" is not produced by this run (needs a cost run); skipped" + ), + } + } + + // `outcome.cells` is index-aligned with `outcome.record.cells` by + // construction: `execute` pushes both in the same cell-loop iteration. + let mut persisted_cells = 0usize; + for (cell_rec, cell_out) in outcome.record.cells.iter().zip(&outcome.cells) { + let Some((_, nominee_report)) = &cell_out.nominee else { + eprintln!( + "aura: cell {}/{}/[{}, {}]: no nominee; no traces persisted", + cell_rec.strategy, cell_rec.instrument, cell_rec.window_ms.0, cell_rec.window_ms.1 + ); + continue; + }; + if routed.is_empty() { + continue; + } + + // The cell key is doc-derived: the window ordinal is the POSITION + // of the cell's window in campaign.data.windows (mirroring the + // family-name convention), never a loop counter re-derived here. + let window_ordinal = campaign + .data + .windows + .iter() + .position(|w| (w.from_ms, w.to_ms) == cell_rec.window_ms) + .ok_or_else(|| { + format!( + "cell window [{}, {}] is not one of campaign.data.windows", + cell_rec.window_ms.0, cell_rec.window_ms.1 + ) + })?; + let cell_key = + campaign_cell_key(&cell_rec.strategy, &cell_rec.instrument, window_ordinal); + + // The cell's blueprint: the run's own index-aligned resolution, by + // content id (exactly the bytes the executor's members ran). + let (_, blueprint_json) = strategies + .iter() + .find(|(id, _)| *id == cell_rec.strategy) + .ok_or_else(|| { + format!( + "cell strategy {} is not among the run's resolved strategies", + cell_rec.strategy + ) + })?; + + // Re-run the nominee, non-reduce: the SAME member the executor ran + // (same wrapped space, same params, same window, seed-free real + // data), mirroring `CliMemberRunner::run_member` with the reduce + // fold off so the per-cycle tap streams exist. The window is the + // nominee report's own `manifest.window` — already epoch-ns + // (`run_blueprint_member` stamped the post-seam bounds), so no + // second ms->ns crossing here. + let space = crate::blueprint_axis_probe(blueprint_json, env).param_space(); + let point = crate::point_from_params(&space, &nominee_report.manifest.params); + let (from, to) = nominee_report.manifest.window; + let no_data = || { + format!( + "no data for instrument {} in the nominee window [{}, {}] (epoch-ns)", + cell_rec.instrument, from.0, to.0 + ) + }; + let source = M1FieldSource::open_window( + server, + &cell_rec.instrument, + Some(from), + Some(to), + M1Field::Close, + ) + .ok_or_else(no_data)?; + if aura_engine::Source::peek(&source).is_none() { + return Err(no_data()); + } + let signal = blueprint_from_json(blueprint_json, &|t| env.resolve(t)) + .expect("stored blueprint passed the referential gate; reload is infallible"); + let (tx_eq, rx_eq) = mpsc::channel(); + let (tx_ex, rx_ex) = mpsc::channel(); + let (tx_r, rx_r) = mpsc::channel(); + let (tx_req, rx_req) = mpsc::channel(); + let mut h = crate::wrap_r(signal, tx_eq, tx_ex, tx_r, tx_req, false, false, None) + .bootstrap_with_cells(&point) + .expect("the nominee's point re-bootstraps (it already ran this realization)"); + let sources: Vec> = vec![Box::new(source)]; + h.run(sources); + // Drain ALL FOUR channels (`run_signal_r` leaves req undrained; the + // trace path must not): eq/ex/req feed the taps, r feeds the metrics. + let eq_rows: Vec<(Timestamp, Vec)> = rx_eq.try_iter().collect(); + let ex_rows: Vec<(Timestamp, Vec)> = rx_ex.try_iter().collect(); + let r_rows: Vec<(Timestamp, Vec)> = rx_r.try_iter().collect(); + let req_rows: Vec<(Timestamp, Vec)> = rx_req.try_iter().collect(); + + // The C1 drift alarm: metrics equality against the recorded + // nominee. The reduce-mode fold shares its arithmetic with this + // non-reduce reduction (SeriesFold via `summarize`; GatedRecorder + // emits exactly the rows `summarize_r`'s ledger reads), so equality + // is bit-exact. + let mut rerun_metrics = summarize(&f64_field(&eq_rows, 0), &f64_field(&ex_rows, 0)); + rerun_metrics.r = Some(summarize_r(&r_rows, &[])); + if rerun_metrics != nominee_report.metrics { + return Err(format!( + "trace re-run diverged from the recorded nominee (C1 violation): cell \ + {cell_key} of trace {trace_name} does not reproduce its recorded \ + metrics; refusing to persist a silently-wrong trace" + )); + } + + let traces: Vec = routed + .iter() + .map(|&(tap, ch)| { + let rows: &[(Timestamp, Vec)] = match ch { + TapChannel::Equity => &eq_rows, + TapChannel::Exposure => &ex_rows, + TapChannel::REquity => &req_rows, + }; + ColumnarTrace::from_rows(tap, &[ScalarKind::F64], rows) + }) + .collect(); + store + .write(&format!("{trace_name}/{cell_key}"), &nominee_report.manifest, &traces) + .map_err(|e| e.to_string())?; + persisted_cells += 1; + } + + eprintln!( + "aura: traces persisted: {trace_name} ({} tap(s) x {persisted_cells} cell(s))", + routed.len() + ); Ok(()) } @@ -591,4 +806,33 @@ mod tests { "raw-namespace prose expected: {m}" ); } + + #[test] + /// 0109: the campaign cell key is content-derived (the `member_key` + /// discipline — never a runtime ordinal): strategy content-id prefix + + /// instrument + doc-positional window ordinal, one portable component. + fn campaign_cell_key_is_content_derived() { + let strategy = "bb34aa55".repeat(8); // a 64-hex content id + assert_eq!(campaign_cell_key(&strategy, "GER40", 0), "bb34aa55-GER40-w0"); + } + + #[test] + /// A non-portable instrument byte maps to `_` (the shared + /// `sanitize_component` charset) — one path component, never a nested + /// path or an invalid directory name. + fn campaign_cell_key_sanitizes_non_portable_bytes() { + let strategy = "0123abcd".repeat(8); + assert_eq!(campaign_cell_key(&strategy, "GER/40", 3), "0123abcd-GER_40-w3"); + } + + #[test] + /// The tap->channel routing mirrors `persist_traces_r` (equity<-eq, + /// exposure<-ex, r_equity<-req); `net_r_equity` is unproducible on the + /// campaign runner's cost-free wrap and routes to no channel. + fn tap_channel_routes_the_producible_vocabulary_only() { + assert_eq!(tap_channel("equity"), Some(TapChannel::Equity)); + assert_eq!(tap_channel("exposure"), Some(TapChannel::Exposure)); + assert_eq!(tap_channel("r_equity"), Some(TapChannel::REquity)); + assert_eq!(tap_channel("net_r_equity"), None); + } } diff --git a/crates/aura-cli/tests/research_docs.rs b/crates/aura-cli/tests/research_docs.rs index b872817..2fe63c2 100644 --- a/crates/aura-cli/tests/research_docs.rs +++ b/crates/aura-cli/tests/research_docs.rs @@ -1344,6 +1344,14 @@ fn campaign_run_valid_tap_reaches_the_member_data_seam() { !out.contains("traces persisted"), "no trace summary on a run that never reached a member: {out}" ); + assert!( + !out.contains("persist_taps not yet honored"), + "the 0107 deferral line is retired; persist_taps is honored from 0109 on: {out}" + ); + assert!( + !out.contains("traces persisted:"), + "a member-data refusal aborts inside execute, before any tracing: {out}" + ); } /// Gated real-data e2e (the `cli_run.rs` skip idiom): a full @@ -1352,8 +1360,13 @@ fn campaign_run_valid_tap_reaches_the_member_data_seam() { /// family/selection lines, a final line parseable as JSON with top-level /// `campaign_run` linking a sweep family id, a walk-forward family id, and a /// pooled-OOS bootstrap on the terminal mc stage, and the -/// `campaign_runs.jsonl` sibling store written. Skips with a note elsewhere -/// so `cargo test --workspace` stays green on a data-less machine. +/// `campaign_runs.jsonl` sibling store written. Since 0109 the campaign +/// also requests `persist_taps: ["equity", "r_equity"]`: the record carries +/// `trace_name`, the nominee's taps land under `traces///` +/// (the persist re-run equality-asserted against the recorded nominee — +/// C1), and `aura chart /` reads them back through the +/// shipped viewer. Skips with a note elsewhere so `cargo test --workspace` +/// stays green on a data-less machine. #[test] fn campaign_run_real_e2e_sweep_gate_walkforward() { let _fixture = project_lock(); @@ -1376,7 +1389,7 @@ fn campaign_run_real_e2e_sweep_gate_walkforward() { &bp_id, &proc_id, (1725148800000, 1727740799999), - "", + "\"equity\", \"r_equity\"", "\"family_table\", \"selection_report\"", ), ); @@ -1448,6 +1461,131 @@ fn campaign_run_real_e2e_sweep_gate_walkforward() { runs_dir.join("campaign_runs.jsonl").is_file(), "campaign_runs.jsonl written beside runs.jsonl" ); + // 0109: the record claims the trace family it persisted under — + // "{campaign8}-{run}" (fresh store => run 0). + let trace_name = v["campaign_run"]["trace_name"] + .as_str() + .expect("persist_taps non-empty => the record carries trace_name"); + assert!( + trace_name.len() == 10 && trace_name.ends_with("-0"), + "trace_name is {{campaign8}}-{{run}}: {trace_name}" + ); + assert!( + trace_name[..8].chars().all(|c| c.is_ascii_hexdigit()), + "trace_name prefix is the campaign id's first 8 hex chars: {trace_name}" + ); + // The summary stderr line names the tap x cell arity (2 requested taps, + // one (strategy, instrument, window) cell). + assert!( + out.contains(&format!("aura: traces persisted: {trace_name} (2 tap(s) x 1 cell(s))")), + "the persist summary line: {out}" + ); + // The taps land under traces/// — the cell key is + // content-derived: strategy8-instrument-w. + let cell_key = format!("{}-GER40-w0", &bp_id[..8]); + let cell_dir = runs_dir.join("traces").join(trace_name).join(&cell_key); + assert!( + cell_dir.join("equity.json").is_file(), + "equity tap persisted at {}", + cell_dir.display() + ); + assert!( + cell_dir.join("r_equity.json").is_file(), + "r_equity tap persisted at {}", + cell_dir.display() + ); + assert!( + !cell_dir.join("exposure.json").exists(), + "unrequested taps are not persisted" + ); + // The shipped viewer reads the member back (the cli_run.rs chart + // idiom): exit 0 and an HTML page carrying both persisted tap series. + let (chart_out, chart_code) = + run_code_in(dir, &["chart", &format!("{trace_name}/{cell_key}")]); + assert_eq!(chart_code, Some(0), "aura chart reads the persisted member: {chart_out}"); + assert!( + chart_out.contains("\"equity\"") && chart_out.contains("\"r_equity\""), + "the chart page carries the two persisted taps: {chart_out}" + ); +} + +/// Property: an UNPRODUCIBLE tap requested alongside a producible one degrades +/// gracefully, never a crash or a silently-dropped whole request. `net_r_equity` +/// is in the closed tap vocabulary (`validate_campaign` accepts it) but has no +/// drained channel on the campaign runner's cost-free wrap (`tap_channel` +/// returns `None`); the one-time stderr note names exactly that tap, the +/// summary line's arity counts only what was actually written (1 tap, not 2), +/// and no `net_r_equity.json` file is ever created — only the producible +/// `equity` tap lands on disk. Uses `SWEEP_ONLY_PROCESS_DOC` (cheapest +/// executable shape) over the same GER40 Sept-2024 window as the gated wf e2e +/// above, so the underlying member run needs real data; skips on a data-less +/// host like its siblings. +#[test] +fn campaign_run_real_e2e_skips_unproducible_tap_gracefully() { + let _fixture = project_lock(); + let dir = built_project(); + let runs_dir = dir.join("runs"); + std::fs::remove_dir_all(&runs_dir).ok(); + let _cleanup = ScratchGuard(vec![ + ScratchPath::Dir(runs_dir.clone()), + ScratchPath::File(dir.join("mixedtap.process.json")), + ScratchPath::File(dir.join("mixedtap.campaign.json")), + ]); + let bp_id = seed_blueprint(dir, "campaign-run-mixedtap-seed"); + let proc_id = register_process_doc(dir, "mixedtap.process.json", SWEEP_ONLY_PROCESS_DOC); + write_doc( + dir, + "mixedtap.campaign.json", + &campaign_doc_json( + &bp_id, + &proc_id, + (1725148800000, 1727740799999), + "\"net_r_equity\", \"equity\"", + "", + ), + ); + let (out, code) = run_code_in(dir, &["campaign", "run", "mixedtap.campaign.json"]); + + // Skip on a data-less machine: the member-data refusal, never a panic. + if code == Some(1) + && (out.contains("no recorded geometry") || out.contains("no data for instrument")) + { + eprintln!("skip: no local GER40 data for the campaign e2e"); + return; + } + + assert_eq!(code, Some(0), "stdout/stderr: {out}"); + let record_line = out + .lines() + .find(|l| l.starts_with("{\"campaign_run\":")) + .expect("the always-on final campaign_run line"); + let v: serde_json::Value = + serde_json::from_str(record_line).expect("campaign_run line parses as JSON"); + let trace_name = v["campaign_run"]["trace_name"] + .as_str() + .expect("persist_taps non-empty => the record carries trace_name"); + + assert!( + out.contains( + "aura: tap \"net_r_equity\" is not produced by this run (needs a cost run); skipped" + ), + "the unproducible tap gets a named, one-time note: {out}" + ); + assert!( + out.contains(&format!("aura: traces persisted: {trace_name} (1 tap(s) x 1 cell(s))")), + "the summary counts only the producible tap actually written: {out}" + ); + let cell_key = format!("{}-GER40-w0", &bp_id[..8]); + let cell_dir = runs_dir.join("traces").join(trace_name).join(&cell_key); + assert!( + cell_dir.join("equity.json").is_file(), + "the producible tap still persists: {}", + cell_dir.display() + ); + assert!( + !cell_dir.join("net_r_equity.json").exists(), + "an unproducible tap is never written to disk" + ); } /// Gated real-data e2e, the OTHER mc bootstrap input shape (#200 d1): with no