Brummel/Aura
1
0
Fork 0
Code Issues 31 Pull Requests Actions Packages Projects Releases Wiki Activity

feat(cli): persist_campaign_traces — campaign presentation persists nominee traces (0109 tasks 3-4)

Browse Source 
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
This commit is contained in:
Brummel
2026-07-04 04:16:22 +02:00
parent b8ef77b1c7
commit 46a85d3ea4
2 changed files with 401 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/aura-cli/src/campaign_run.rs
+260 -16
View File
@@ -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<TapChannel> {
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/<trace_name>/` — 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/<trace_name>/<cell_key>/` (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<data_server::DataServer>,
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<Box<dyn aura_engine::Source>> = 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<Scalar>)> = rx_eq.try_iter().collect();
let ex_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_ex.try_iter().collect();
let r_rows: Vec<(Timestamp, Vec<Scalar>)> = rx_r.try_iter().collect();
let req_rows: Vec<(Timestamp, Vec<Scalar>)> = 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<ColumnarTrace> = routed
.iter()
.map(|&(tap, ch)| {
let rows: &[(Timestamp, Vec<Scalar>)] = 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);
}
}
crates/aura-cli/tests/research_docs.rs
+141 -3
View File
@@ -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/<name>/<cell_key>/`
/// (the persist re-run equality-asserted against the recorded nominee —
/// C1), and `aura chart <name>/<cell_key>` 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/<name>/<cell_key>/ — the cell key is
// content-derived: strategy8-instrument-w<window_ordinal>.
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