diff --git a/Cargo.lock b/Cargo.lock index e2afc99..44d3cec 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -122,6 +122,7 @@ dependencies = [ name = "aura-cli" version = "0.1.0" dependencies = [ + "aura-campaign", "aura-composites", "aura-core", "aura-engine", diff --git a/crates/aura-campaign/src/exec.rs b/crates/aura-campaign/src/exec.rs index 7cea624..2e46410 100644 --- a/crates/aura-campaign/src/exec.rs +++ b/crates/aura-campaign/src/exec.rs @@ -97,7 +97,9 @@ pub fn execute( // The store's self-keying always yields a 64-hex content id; anything else // is a caller bug — refuse it typed instead of panicking on the prefix // slice below. - if campaign_id.len() != 64 || !campaign_id.bytes().all(|b| b.is_ascii_hexdigit()) { + if campaign_id.len() != 64 + || !campaign_id.bytes().all(|b| matches!(b, b'0'..=b'9' | b'a'..=b'f')) + { return Err(ExecFault::PipelineShape { detail: format!( "campaign_id must be a 64-hex content id (got {} chars)", diff --git a/crates/aura-cli/Cargo.toml b/crates/aura-cli/Cargo.toml index 2f5b8bf..d6d4d61 100644 --- a/crates/aura-cli/Cargo.toml +++ b/crates/aura-cli/Cargo.toml @@ -17,6 +17,10 @@ aura-engine = { path = "../aura-engine" } aura-composites = { path = "../aura-composites" } aura-registry = { path = "../aura-registry" } aura-research = { path = "../aura-research" } +# aura-campaign: campaign-execution semantics (preflight, cell loop, stage +# sequencing, realization record); the CLI implements its MemberRunner seam +# and renders its outcome (#198). +aura-campaign = { path = "../aura-campaign" } aura-std = { path = "../aura-std" } aura-ingest = { path = "../aura-ingest" } # data-server: the local M1 archive `aura run --real` streams from. Mirrors the diff --git a/crates/aura-cli/src/campaign_run.rs b/crates/aura-cli/src/campaign_run.rs new file mode 100644 index 0000000..45c55f3 --- /dev/null +++ b/crates/aura-cli/src/campaign_run.rs @@ -0,0 +1,493 @@ +//! `aura campaign run` — the driver that turns a stored campaign document into +//! a realized run-set (#198). The execution *semantics* (preflight, cell loop, +//! stage sequencing, selection, registry writes) live in `aura-campaign`; this +//! module owns what is CLI-specific: target resolution (file sugar vs content +//! id), the project + referential gates, the [`MemberRunner`] implementation +//! over the shipped loaded-blueprint machinery (`wrap_r` reduce-mode member +//! runs via `run_blueprint_member` + `M1FieldSource` windowed real-data +//! binding), fault prose (`exec_fault_prose` lives HERE, beside its consumer, +//! not in `research_docs.rs` — it phrases `aura_campaign` types the doc-tier +//! module deliberately does not import), and stdout/stderr emission. +//! +//! Root items (`blueprint_axis_probe`, `run_blueprint_member`, +//! `family_member_line`) are reached via `crate::` — the `graph_construct` +//! submodule idiom: main.rs is the crate root, so its private items are +//! visible to child modules without promotion. + +use std::path::{Path, PathBuf}; +use std::sync::Arc; + +use aura_campaign::{CellSpec, ExecFault, MemberFault, MemberRunner}; +use aura_core::{Cell, ParamSpec, Scalar}; +use aura_engine::{blueprint_from_json, FamilySelection, RunReport}; +use aura_ingest::{instrument_geometry, unix_ms_to_epoch_ns, M1Field, M1FieldSource}; +use aura_registry::CampaignRunRecord; +use aura_research::{ + campaign_to_json, content_id_of, parse_campaign, parse_process, validate_campaign, + validate_process, DocRef, +}; + +use crate::project::Env; +use crate::research_docs::{ + doc_error_prose, doc_fault_prose, fault_block, parse_valid_campaign, ref_fault_prose, +}; + +/// A bare store address: exactly 64 lowercase hex chars (the content-id key shape). +fn is_content_id(s: &str) -> bool { + s.len() == 64 && s.bytes().all(|b| matches!(b, b'0'..=b'9' | b'a'..=b'f')) +} + +/// Phrase an [`ExecFault`] for stderr — Debug-leak-free, path-addressed +/// (stage index + block id), the `doc_fault_prose`/`ref_fault_prose` register. +fn exec_fault_prose(f: &ExecFault) -> String { + match f { + ExecFault::UnsupportedStage { stage, block } => format!( + "process stage {stage} ({block}) is not executable in v1; executable \ + pipeline shape: std::sweep [std::gate]* [std::walk_forward]" + ), + ExecFault::PipelineShape { detail } => { + format!("process pipeline is not executable: {detail}") + } + ExecFault::UnrankableMetric { stage, metric } => format!( + "process stage {stage}: metric \"{metric}\" is not rankable (winner \ + selection needs one of the registry's rankable metrics)" + ), + ExecFault::GateMetricNotPerMember { stage, metric } => format!( + "process stage {stage}: gate metric \"{metric}\" is not a per-member \ + scalar (selection annotations cannot gate members)" + ), + ExecFault::PlateauInWalkForward { stage } => format!( + "process stage {stage}: walk_forward cannot use a plateau select (a \ + gated survivor subset has no parameter lattice)" + ), + ExecFault::DeflatePlateauConflict { stage } => format!( + "process stage {stage}: sweep deflate: true composes only with select \"argmax\"" + ), + ExecFault::Window { stage, detail } => format!( + "process stage {stage}: walk_forward windows do not fit the campaign \ + window: {detail}" + ), + ExecFault::Member(MemberFault::NoData { instrument, window_ms }) => format!( + "no data for instrument {instrument} in window [{}, {}] (epoch-ms)", + window_ms.0, window_ms.1 + ), + ExecFault::Member(MemberFault::Bind(detail)) => { + format!("a member failed to bind: {detail}") + } + ExecFault::Member(MemberFault::Run(detail)) => { + format!("a member failed to run: {detail}") + } + ExecFault::Registry(e) => e.to_string(), + } +} + +/// stdout wire form of one selection-bearing stage. Field order is the wire +/// contract (serde derives declaration order). +#[derive(serde::Serialize)] +struct SelectionReportLine<'a> { + selection_report: SelectionReportBody<'a>, +} + +#[derive(serde::Serialize)] +struct SelectionReportBody<'a> { + family_id: &'a str, + stage: usize, + block: &'a str, + winner_ordinal: usize, + params: &'a [(String, Scalar)], + selection: &'a FamilySelection, +} + +/// The always-on final line: the stored realization record under one key. +#[derive(serde::Serialize)] +struct CampaignRunLine<'a> { + campaign_run: &'a CampaignRunRecord, +} + +/// Suffix-join each raw campaign axis onto exactly one wrapped param (wrapped +/// == raw, or wrapped ends with ".{raw}" — the established suffix-match +/// pattern), then require every wrapped slot to be covered. Pure and +/// independent of the env/data seam so the three fault arms are unit-testable +/// without a project, a store, or a loaded blueprint. +fn bind_axes( + space: &[ParamSpec], + strategy_id: &str, + params: &[(String, Scalar)], +) -> Result, MemberFault> { + let mut slots: Vec> = vec![None; space.len()]; + for (raw, value) in params { + let hits: Vec = space + .iter() + .enumerate() + .filter(|(_, p)| p.name == *raw || p.name.ends_with(&format!(".{raw}"))) + .map(|(i, _)| i) + .collect(); + match hits.as_slice() { + [i] => slots[*i] = Some(*value), + [] => { + return Err(MemberFault::Bind(format!( + "axis \"{raw}\" matches no open param of the wrapped strategy {strategy_id}" + ))); + } + _ => { + let names: Vec<&str> = hits.iter().map(|&i| space[i].name.as_str()).collect(); + return Err(MemberFault::Bind(format!( + "axis \"{raw}\" is ambiguous in the wrapped param space of \ + strategy {strategy_id}: matches {}", + names.join(", ") + ))); + } + } + } + let mut point: Vec = Vec::with_capacity(space.len()); + for (spec, slot) in space.iter().zip(&slots) { + match slot { + Some(v) => point.push(v.cell()), + None => { + return Err(MemberFault::Bind(format!( + "open param \"{}\" of strategy {strategy_id} is bound by no campaign axis", + spec.name + ))); + } + } + } + Ok(point) +} + +/// The CLI's harness/data binding seam for `aura_campaign::execute`: members +/// run through the shipped loaded-blueprint machinery (`wrap_r` reduce-mode +/// via `crate::run_blueprint_member`) over windowed real M1 close bars +/// (`M1FieldSource::open_window` — the ms→ns crossing happens at exactly this +/// seam, via `unix_ms_to_epoch_ns`). All refusals are member faults for the +/// library to surface; never a process exit inside a sweep worker. +struct CliMemberRunner<'a> { + env: &'a Env, + server: Arc, +} + +impl MemberRunner for CliMemberRunner<'_> { + fn run_member( + &self, + cell: &CellSpec, + params: &[(String, Scalar)], + window_ms: (i64, i64), + ) -> Result { + // The wrapped axis namespace — the SAME probe the sweep verbs resolve + // against (single-sourced in main.rs; identical `false, true, None` wrap). + let space = crate::blueprint_axis_probe(&cell.blueprint_json, self.env).param_space(); + let point = bind_axes(&space, &cell.strategy_id, params)?; + + // Real windowed data — geometry BEFORE bar data (the shipped pre-data + // refusal order of `open_real_source`), both as member faults. + let geo = instrument_geometry(&self.server, &cell.instrument).ok_or_else(|| { + MemberFault::Run(format!( + "no recorded geometry for symbol '{}' at {} — refusing to run a \ + real instrument with a guessed pip", + cell.instrument, + self.env.data_path() + )) + })?; + let from = unix_ms_to_epoch_ns(window_ms.0); + let to = unix_ms_to_epoch_ns(window_ms.1); + let source = match M1FieldSource::open_window( + &self.server, + &cell.instrument, + Some(from), + Some(to), + M1Field::Close, + ) { + Some(s) => s, + // No archived file overlaps the window at all. + None => { + return Err(MemberFault::NoData { + instrument: cell.instrument.clone(), + window_ms, + }); + } + }; + // A window that overlaps a file but holds zero matching bars yields a + // source whose first peek is None (open_window's documented contract) + // — the same no-data condition. + if aura_engine::Source::peek(&source).is_none() { + return Err(MemberFault::NoData { + instrument: cell.instrument.clone(), + window_ms, + }); + } + + // The shipped member recipe (reload — a Composite is !Clone — wrap, + // bind, run, summarize): `run_blueprint_member` verbatim. Seed 0 + // (seed-free real-data runs), topology_hash = the strategy's content + // id, project provenance stamped inside the helper. + let signal = blueprint_from_json(&cell.blueprint_json, &|t| self.env.resolve(t)) + .expect("stored blueprint passed the referential gate; reload is infallible"); + let mut report = crate::run_blueprint_member( + signal, + &point, + &space, + vec![Box::new(source)], + (from, to), + 0, + geo.pip_size, + &cell.strategy_id, + self.env, + ); + report.manifest.instrument = Some(cell.instrument.clone()); + Ok(report) + } +} + +/// `aura campaign run `: resolve, gate, execute, emit. Every `Err` +/// is a refusal `campaign_cmd` renders as `aura: {msg}` + exit 1. +pub(crate) fn run_campaign(target: &str, env: &Env) -> Result<(), String> { + // Project gate FIRST: nothing (not even the file-sugar registration) + // touches a store outside a project. + if env.provenance().is_none() { + let cwd = std::env::current_dir() + .map(|d| d.display().to_string()) + .unwrap_or_default(); + return Err(format!( + "campaign run needs a project: strategies resolve against the project \ + store and vocabulary (no Aura.toml found up from {cwd})" + )); + } + let registry = env.registry(); + + // Target resolution: a readable file is register-then-run sugar; a bare + // 64-hex token addresses the store directly; anything else refuses naming + // both readings. + let campaign_id = if Path::new(target).is_file() { + let doc = parse_valid_campaign(&PathBuf::from(target))?; + registry + .put_campaign(&campaign_to_json(&doc)) + .map_err(|e| e.to_string())? + } else if is_content_id(target) { + target.to_string() + } else { + return Err(format!( + "'{target}' is neither a readable .json file nor a 64-hex content id" + )); + }; + + // One uniform path from here: fetch the stored canonical bytes by id (so + // file addressing and id addressing produce the same realization by + // construction) and re-run the intrinsic tier on them. + let campaign_text = registry + .get_campaign(&campaign_id) + .map_err(|e| e.to_string())? + .ok_or_else(|| format!("no campaign {campaign_id} in the project store"))?; + let campaign = parse_campaign(&campaign_text) + .map_err(|e| doc_error_prose("stored campaign document", &e))?; + let faults = validate_campaign(&campaign); + if !faults.is_empty() { + return Err(fault_block( + "campaign document invalid:", + faults.iter().map(doc_fault_prose).collect(), + )); + } + + // Referential gate: zero faults or refuse (the campaign-validate seam). + let resolve = |t: &str| env.resolve(t); + let ref_faults = registry + .validate_campaign_refs(&campaign, &resolve) + .map_err(|e| e.to_string())?; + if !ref_faults.is_empty() { + return Err(fault_block( + "campaign references do not resolve:", + ref_faults.iter().map(ref_fault_prose).collect(), + )); + } + + // Process fetch + intrinsic validation (stored text, not a file path — + // parse_valid_process is file-based, so its constituents run here). + let DocRef::ContentId(process_id) = &campaign.process.r#ref else { + // validate_campaign already refuses identity process refs; defensive. + return Err("process.ref: a process is referenced by content id in this version".into()); + }; + let process_text = registry + .get_process(process_id) + .map_err(|e| e.to_string())? + .ok_or_else(|| format!("no process {process_id} in the project store"))?; + let process = parse_process(&process_text) + .map_err(|e| doc_error_prose("stored process document", &e))?; + let process_faults = validate_process(&process); + if !process_faults.is_empty() { + return Err(fault_block( + "process document invalid:", + process_faults.iter().map(doc_fault_prose).collect(), + )); + } + + // Strategies: canonical bytes from the store, index-aligned with the doc. + // The recorded id is the content id of those bytes (== the ref id for + // content refs; computed for identity refs). + let mut strategies: Vec<(String, String)> = Vec::with_capacity(campaign.strategies.len()); + for entry in &campaign.strategies { + let canonical = match &entry.r#ref { + DocRef::ContentId(id) => registry + .get_blueprint(id) + .map_err(|e| e.to_string())? + .ok_or_else(|| format!("strategy {id} not found in the blueprint store"))?, + DocRef::IdentityId(id) => registry + .find_blueprint_by_identity(id, &resolve) + .map_err(|e| e.to_string())? + .ok_or_else(|| format!("identity id {id} matches no stored blueprint"))?, + }; + 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())), + }; + let outcome = aura_campaign::execute( + &campaign_id, + &campaign, + &process, + &strategies, + &runner, + ®istry, + ) + .map_err(|f| exec_fault_prose(&f))?; + + // Zero-survivor stderr notes (exit stays 0 — a null result is a valid + // research result, #198 decision 8). Addressed by the fields the record + // already carries (strategy/instrument/window_ms), not by re-deriving a + // doc-order ordinal from the loop-nesting the executor happens to use + // today — that duplicated invariant would silently drift if the executor + // ever reorders its cell loop. + for cell in &outcome.record.cells { + for (stage_ix, st) in cell.stages.iter().enumerate() { + if matches!(&st.survivor_ordinals, Some(v) if v.is_empty()) { + eprintln!( + "aura: cell {}/{}/[{}, {}]: gate at stage {stage_ix} left no \ + survivors; cell realization truncated", + cell.strategy, cell.instrument, cell.window_ms.0, cell.window_ms.1 + ); + } + } + } + + // Emission: emit-gated family/selection lines per cell, then the + // always-on final record line. Matched by NAME against the two closed- + // vocabulary terms (self-evident at the call site, unlike positional + // indexing into the vocab slice); the debug_assert keeps the names + // honest against `aura_research::emit_vocabulary()` so a #190 + // rename/extend of the closed set fails loudly here instead of silently + // misrouting emission. + debug_assert!( + aura_research::emit_vocabulary().contains(&"family_table") + && aura_research::emit_vocabulary().contains(&"selection_report"), + "emit_vocabulary drifted from the names campaign_run matches by" + ); + let emit_family = campaign.presentation.emit.iter().any(|e| e == "family_table"); + let emit_selection = campaign.presentation.emit.iter().any(|e| e == "selection_report"); + for cell_out in &outcome.cells { + if emit_family { + for fam in &cell_out.families { + for report in &fam.reports { + println!("{}", crate::family_member_line(&fam.family_id, report)); + } + } + } + if emit_selection { + for sel in &cell_out.selections { + let line = SelectionReportLine { + selection_report: SelectionReportBody { + family_id: &sel.family_id, + stage: sel.stage, + block: sel.block, + winner_ordinal: sel.winner_ordinal, + params: &sel.params, + selection: &sel.selection, + }, + }; + println!( + "{}", + serde_json::to_string(&line).expect("selection report serializes") + ); + } + } + } + println!( + "{}", + serde_json::to_string(&CampaignRunLine { campaign_run: &outcome.record }) + .expect("campaign run record serializes") + ); + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + use aura_core::ScalarKind; + + fn spec(name: &str) -> ParamSpec { + ParamSpec { name: name.to_string(), kind: ScalarKind::I64 } + } + + #[test] + /// The only shape treated as a direct store address is a bare 64-char + /// lowercase-hex token; anything else (wrong length, uppercase, non-hex) + /// falls through to `run_campaign`'s file-path branch instead. + fn is_content_id_accepts_only_64_lowercase_hex() { + assert!(is_content_id(&"a".repeat(64))); + assert!(!is_content_id(&"A".repeat(64))); + assert!(!is_content_id(&"a".repeat(63))); + assert!(!is_content_id(&format!("{}g", "a".repeat(63)))); + } + + #[test] + /// bind_axes resolves a raw axis name against the ONE wrapped param whose + /// path ends with ".{raw}" (or equals it), producing a co-indexed point. + fn bind_axes_resolves_a_unique_suffix_match() { + let space = vec![spec("sma.length")]; + let params = vec![("length".to_string(), Scalar::i64(14))]; + let point = bind_axes(&space, "strat", ¶ms).unwrap(); + assert_eq!(point, vec![Scalar::i64(14).cell()]); + } + + #[test] + /// A raw campaign axis that matches no open param of the wrapped + /// strategy is a named Bind fault naming the axis, never a silently + /// dropped binding. + fn bind_axes_refuses_an_unmatched_axis() { + let space = vec![spec("sma.length")]; + let params = vec![("period".to_string(), Scalar::i64(14))]; + let err = bind_axes(&space, "strat", ¶ms).unwrap_err(); + let MemberFault::Bind(m) = err else { panic!("expected Bind, got {err:?}") }; + assert!(m.contains("period") && m.contains("matches no open param")); + } + + #[test] + /// A raw axis matching MORE THAN ONE wrapped param is a named Bind fault + /// that lists every ambiguous candidate, never a silent first-match pick. + fn bind_axes_refuses_an_ambiguous_axis() { + let space = vec![spec("fast.length"), spec("slow.length")]; + let params = vec![("length".to_string(), Scalar::i64(14))]; + let err = bind_axes(&space, "strat", ¶ms).unwrap_err(); + let MemberFault::Bind(m) = err else { panic!("expected Bind, got {err:?}") }; + assert!(m.contains("ambiguous") && m.contains("fast.length") && m.contains("slow.length")); + } + + #[test] + /// An open wrapped param left unbound by any campaign axis is a named + /// Bind fault naming the param, not an implicit default. + fn bind_axes_refuses_an_uncovered_param() { + let space = vec![spec("sma.length"), spec("sma.threshold")]; + let params = vec![("length".to_string(), Scalar::i64(14))]; + let err = bind_axes(&space, "strat", ¶ms).unwrap_err(); + let MemberFault::Bind(m) = err else { panic!("expected Bind, got {err:?}") }; + assert!(m.contains("threshold") && m.contains("bound by no campaign axis")); + } +} diff --git a/crates/aura-cli/src/main.rs b/crates/aura-cli/src/main.rs index 662ebaa..c3f073f 100644 --- a/crates/aura-cli/src/main.rs +++ b/crates/aura-cli/src/main.rs @@ -14,6 +14,7 @@ mod render; mod graph_construct; mod project; +mod campaign_run; mod research_docs; mod scaffold; use render::{ChartData, ChartMeta, ChartMode, ReduceKind, Series}; diff --git a/crates/aura-cli/src/research_docs.rs b/crates/aura-cli/src/research_docs.rs index 700303c..9b93e7b 100644 --- a/crates/aura-cli/src/research_docs.rs +++ b/crates/aura-cli/src/research_docs.rs @@ -67,7 +67,7 @@ fn guard_one_mode(cmd: &DocIntrospectCmd, family: &str) { } } -fn doc_error_prose(what: &str, e: &DocError) -> String { +pub(crate) fn doc_error_prose(what: &str, e: &DocError) -> String { match e { DocError::NotJson(msg) => format!("{what} is not JSON: {msg}"), DocError::BadFormatVersion(v) => { @@ -84,7 +84,7 @@ fn doc_error_prose(what: &str, e: &DocError) -> String { } } -fn doc_fault_prose(f: &DocFault) -> String { +pub(crate) fn doc_fault_prose(f: &DocFault) -> String { match f { DocFault::EmptyPipeline => "the pipeline is empty — a process needs at least one stage".into(), DocFault::UnknownMetric { stage, metric } => { @@ -121,7 +121,7 @@ fn read_doc(file: &PathBuf) -> Result { std::fs::read_to_string(file).map_err(|e| format!("cannot read {}: {e}", file.display())) } -fn fault_block(header: &str, lines: Vec) -> String { +pub(crate) fn fault_block(header: &str, lines: Vec) -> String { format!("{header}\n {}", lines.join("\n ")) } @@ -244,9 +244,12 @@ pub enum CampaignSub { Introspect(DocIntrospectCmd), /// Register a valid campaign document into the store under the runs root. Register { file: PathBuf }, + /// Execute a stored campaign into a realized run-set (a .json file is + /// register-then-run sugar; the canonical address is the content id). + Run { target: String }, } -fn ref_fault_prose(f: &RefFault) -> String { +pub(crate) fn ref_fault_prose(f: &RefFault) -> String { match f { RefFault::ProcessNotFound(id) => format!("process {id} not found in the project store"), RefFault::StrategyNotFound(id) => format!("strategy {id} not found in the blueprint store"), @@ -272,6 +275,7 @@ pub fn campaign_cmd(cmd: CampaignCmd, env: &Env) { introspect_campaign(i) } CampaignSub::Register { file } => register_campaign(file, env), + CampaignSub::Run { target } => crate::campaign_run::run_campaign(target, env), }; if let Err(m) = result { eprintln!("aura: {m}"); @@ -279,7 +283,7 @@ pub fn campaign_cmd(cmd: CampaignCmd, env: &Env) { } } -fn parse_valid_campaign(file: &PathBuf) -> Result { +pub(crate) fn parse_valid_campaign(file: &PathBuf) -> Result { let text = read_doc(file)?; let doc = parse_campaign(&text).map_err(|e| doc_error_prose("campaign document", &e))?; let faults = validate_campaign(&doc); diff --git a/crates/aura-cli/tests/research_docs.rs b/crates/aura-cli/tests/research_docs.rs index b9dff65..b4df9c6 100644 --- a/crates/aura-cli/tests/research_docs.rs +++ b/crates/aura-cli/tests/research_docs.rs @@ -3,7 +3,7 @@ //! author uses. use std::path::{Path, PathBuf}; -use std::sync::OnceLock; +use std::sync::{Mutex, MutexGuard, OnceLock}; /// A fresh, unique working directory for a process test that persists /// content-addressed documents under `./runs/` (so `aura process register` @@ -217,6 +217,15 @@ fn built_project() -> &'static PathBuf { }) } +/// Serializes every test that touches the shared demo-project fixture store +/// (they remove/re-seed `/runs`, so parallel test threads would race +/// on it). A poisoned lock is taken over: one failed test must not cascade +/// into unrelated lock panics. +fn project_lock() -> MutexGuard<'static, ()> { + static LOCK: Mutex<()> = Mutex::new(()); + LOCK.lock().unwrap_or_else(|e| e.into_inner()) +} + /// A scratch filesystem entry this test writes under the git-tracked /// demo-project fixture root (only `runs/` there is fixture-gitignored), /// removed on drop — including during a mid-test panic — so a failed @@ -253,6 +262,7 @@ impl Drop for ScratchGuard { /// them together end to end. #[test] fn campaign_validate_in_project_reports_referential_tier_end_to_end() { + let _fixture = project_lock(); let dir = built_project(); let runs_dir = dir.join("runs"); std::fs::remove_dir_all(&runs_dir).ok(); @@ -477,3 +487,388 @@ fn campaign_register_refuses_invalid_document_and_writes_nothing() { "register must not create a store entry for an invalid document" ); } + +/// Seed one open-param blueprint into the built demo project's store via a +/// real sweep and return its content id (the referential test's recipe). +fn seed_blueprint(dir: &Path, name: &str) -> String { + let open_bp = format!("{}/tests/fixtures/sma_signal_open.json", env!("CARGO_MANIFEST_DIR")); + let (out, code) = run_code_in( + dir, + &[ + "sweep", + &open_bp, + "--axis", + "sma_signal.fast.length=2,4", + "--axis", + "sma_signal.slow.length=8,16", + "--name", + name, + ], + ); + assert_eq!(code, Some(0), "seed sweep failed: {out}"); + std::fs::read_dir(dir.join("runs").join("blueprints")) + .expect("blueprints dir") + .next() + .expect("one stored blueprint") + .expect("dir entry") + .path() + .file_stem() + .expect("stem") + .to_string_lossy() + .into_owned() +} + +/// Register `doc` as a process document in the project store; returns its id. +/// Asserts register exits 0 — an intrinsically valid document (an mc-bearing +/// one included) must always register; only `campaign run` draws the v1 line. +fn register_process_doc(dir: &Path, file: &str, doc: &str) -> String { + write_doc(dir, file, doc); + let (out, code) = run_code_in(dir, &["process", "register", file]); + assert_eq!(code, Some(0), "process register failed: {out}"); + out.lines() + .find(|l| l.starts_with("registered process content:")) + .expect("register line") + .trim_start_matches("registered process content:") + .split(' ') + .next() + .expect("id") + .to_string() +} + +/// A referentially-resolving campaign over the seeded blueprint. Axes name +/// the RAW `param_space` names (`fast.length` / `slow.length` — see the +/// naming note in the referential test above). `persist_taps`/`emit` are +/// spliced verbatim (pass `""` for empty, `"\"family_table\""` etc.). +fn campaign_doc_json( + bp_id: &str, + proc_id: &str, + window: (i64, i64), + persist_taps: &str, + emit: &str, +) -> String { + format!( + r#"{{ + "format_version": 1, + "kind": "campaign", + "name": "run-seam", + "data": {{ "instruments": ["GER40"], "windows": [ {{ "from_ms": {from}, "to_ms": {to} }} ] }}, + "strategies": [ {{ "ref": {{ "content_id": "{bp_id}" }}, + "axes": {{ "fast.length": {{ "kind": "I64", "values": [2, 4] }}, + "slow.length": {{ "kind": "I64", "values": [8, 16] }} }} }} ], + "process": {{ "ref": {{ "content_id": "{proc_id}" }} }}, + "seed": 7, + "presentation": {{ "persist_taps": [{persist_taps}], "emit": [{emit}] }} +}}"#, + from = window.0, + to = window.1, + ) +} + +/// An mc-bearing process: intrinsically VALID (register accepts it) but past +/// the executable v1 boundary (`campaign run` refuses it at preflight). +const MC_PROCESS_DOC: &str = r#"{ + "format_version": 1, + "kind": "process", + "name": "mc-screen", + "pipeline": [ + { "block": "std::sweep", "metric": "sqn_normalized", "select": "argmax" }, + { "block": "std::monte_carlo", "resamples": 100, "block_len": 5 } + ] +}"#; + +/// The minimal executable v1 pipeline (one sweep stage). +const SWEEP_ONLY_PROCESS_DOC: &str = r#"{ + "format_version": 1, + "kind": "process", + "name": "sweep-only", + "pipeline": [ { "block": "std::sweep", "metric": "sqn_normalized", "select": "argmax" } ] +}"#; + +/// The full v1 shape for the gated e2e: sweep -> gate -> walk_forward. The +/// gate (`n_trades ge 0`) passes every member, so walk-forward always has +/// survivors. Roller: 14d IS / 7d OOS / 7d step in epoch-ms, tiling the +/// ~30-day GER40 Sept-2024 campaign window. +const WF_PROCESS_DOC: &str = r#"{ + "format_version": 1, + "kind": "process", + "name": "screen-then-walkforward", + "pipeline": [ + { "block": "std::sweep", "metric": "sqn_normalized", "select": "argmax" }, + { "block": "std::gate", "all": [ { "metric": "n_trades", "cmp": "ge", "value": 0.0 } ] }, + { "block": "std::walk_forward", "in_sample_ms": 1209600000, "out_of_sample_ms": 604800000, + "step_ms": 604800000, "mode": "rolling", "metric": "net_expectancy_r", "select": "argmax" } + ] +}"#; + +/// `campaign run` outside a project refuses up front — before target +/// resolution, so not even the file-sugar registration touches a store. +#[test] +fn campaign_run_outside_project_refuses() { + let dir = temp_cwd("campaign-run-outside-project"); + write_doc(&dir, "c.campaign.json", CAMPAIGN_DOC); + let (out, code) = run_code_in(&dir, &["campaign", "run", "c.campaign.json"]); + assert_eq!(code, Some(1), "stdout/stderr: {out}"); + assert!(out.contains("campaign run needs a project"), "stdout/stderr: {out}"); + assert!( + !dir.join("runs").exists(), + "a refused run must not create a store outside a project" + ); +} + +/// A target that is neither a readable file nor a 64-hex id refuses naming +/// both readings (inside the project, past the project gate). +#[test] +fn campaign_run_bogus_target_refuses() { + let _fixture = project_lock(); + let dir = built_project(); + let (out, code) = run_code_in(dir, &["campaign", "run", "no-such-target"]); + assert_eq!(code, Some(1), "stdout/stderr: {out}"); + assert!( + out.contains("'no-such-target' is neither a readable .json file nor a 64-hex content id"), + "stdout/stderr: {out}" + ); +} + +/// A well-formed but unknown content id refuses with the store-miss prose. +#[test] +fn campaign_run_unknown_id_refuses() { + let _fixture = project_lock(); + let dir = built_project(); + let id = "0".repeat(64); + let (out, code) = run_code_in(dir, &["campaign", "run", &id]); + assert_eq!(code, Some(1), "stdout/stderr: {out}"); + assert!( + out.contains(&format!("no campaign {id} in the project store")), + "stdout/stderr: {out}" + ); +} + +/// The v1 boundary: `process register` ACCEPTS an mc-bearing document (it is +/// intrinsically valid — asserted inside `register_process_doc`); only +/// `campaign run` refuses it, at preflight, before any member runs (so no +/// data is needed), with path-addressed Debug-free prose. +#[test] +fn campaign_run_v1_boundary_refuses_mc_process() { + 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("mc.process.json")), + ScratchPath::File(dir.join("mc.campaign.json")), + ]); + let bp_id = seed_blueprint(dir, "campaign-run-mc-seed"); + let proc_id = register_process_doc(dir, "mc.process.json", MC_PROCESS_DOC); + write_doc(dir, "mc.campaign.json", &campaign_doc_json(&bp_id, &proc_id, (1, 2), "", "")); + let (out, code) = run_code_in(dir, &["campaign", "run", "mc.campaign.json"]); + assert_eq!(code, Some(1), "stdout/stderr: {out}"); + assert!(out.contains("not executable in v1"), "stdout/stderr: {out}"); + assert!(out.contains("std::monte_carlo"), "the prose names the block: {out}"); + assert!(!out.contains("UnsupportedStage"), "Debug leak: {out}"); +} + +/// Non-empty persist_taps defers LOUDLY, and the note's ORDER is pinned: it +/// prints before member execution, so it is asserted here on a run +/// that refuses at the member-data seam. The [1, 2] epoch-ms window (1970) +/// makes that refusal deterministic on every machine: a data-less host +/// refuses on missing geometry, a data-ful host on a window no archive file +/// overlaps — exit 1 either way, tap note already on stderr. +#[test] +fn campaign_run_persist_taps_deferred_loudly() { + 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("taps.process.json")), + ScratchPath::File(dir.join("taps.campaign.json")), + ]); + let bp_id = seed_blueprint(dir, "campaign-run-taps-seed"); + let proc_id = register_process_doc(dir, "taps.process.json", SWEEP_ONLY_PROCESS_DOC); + write_doc( + dir, + "taps.campaign.json", + &campaign_doc_json(&bp_id, &proc_id, (1, 2), "\"r_record\"", ""), + ); + let (out, code) = run_code_in(dir, &["campaign", "run", "taps.campaign.json"]); + assert_eq!(code, Some(1), "stdout/stderr: {out}"); + assert!( + out.contains("aura: persist_taps not yet honored (1 tap(s) ignored)"), + "the tap note must precede the member-data refusal: {out}" + ); + assert!( + out.contains("no recorded geometry") || out.contains("no data for instrument"), + "the refusal names the data condition: {out}" + ); +} + +/// Gated real-data e2e (the `cli_run.rs` skip idiom): a full +/// sweep -> gate -> walk_forward campaign over GER40 Sept-2024 where the +/// local archive is present — exit 0, emit-gated family/selection lines, a +/// final line parseable as JSON with top-level `campaign_run` linking a sweep +/// family id and a walk-forward family id, and the `campaign_runs.jsonl` +/// sibling store written. 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(); + 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("wf.process.json")), + ScratchPath::File(dir.join("wf.campaign.json")), + ]); + let bp_id = seed_blueprint(dir, "campaign-run-e2e-seed"); + let proc_id = register_process_doc(dir, "wf.process.json", WF_PROCESS_DOC); + // The GER40 Sept-2024 window (inclusive Unix-ms) — the same gated window + // cli_run.rs drives; ~30 days, so the (14d, 7d, 7d) roller tiles it. + write_doc( + dir, + "wf.campaign.json", + &campaign_doc_json( + &bp_id, + &proc_id, + (1725148800000, 1727740799999), + "", + "\"family_table\", \"selection_report\"", + ), + ); + let (out, code) = run_code_in(dir, &["campaign", "run", "wf.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 cells = v["campaign_run"]["cells"].as_array().expect("cells array"); + assert_eq!(cells.len(), 1, "one (strategy, instrument, window) cell: {record_line}"); + let stages = cells[0]["stages"].as_array().expect("stages array"); + assert_eq!(stages.len(), 3, "sweep + gate + walk_forward realized: {record_line}"); + assert!( + stages[0]["family_id"].as_str().is_some(), + "sweep stage links a family: {record_line}" + ); + assert_eq!( + stages[1]["survivor_ordinals"].as_array().map(|a| a.len()), + Some(4), + "the always-true gate keeps all four members: {record_line}" + ); + assert!( + stages[2]["family_id"].as_str().is_some(), + "walk-forward stage links a family: {record_line}" + ); + // Emit-gated lines: per-member family_table lines (4 sweep members plus + // the walk-forward OOS members) and at least one selection_report line. + assert!( + out.lines().filter(|l| l.starts_with("{\"family_id\":")).count() >= 4, + "family_table member lines emitted: {out}" + ); + assert!( + out.lines().any(|l| l.starts_with("{\"selection_report\":")), + "selection_report line emitted: {out}" + ); + // The registry's new sibling store carries the realization record. + assert!( + runs_dir.join("campaign_runs.jsonl").is_file(), + "campaign_runs.jsonl written beside runs.jsonl" + ); +} + +/// `run_campaign` resolves a target two ways — a readable file (register-then- +/// run sugar) or a bare content id (direct store address) — then funnels BOTH +/// through one shared post-resolution path: "fetch the stored canonical bytes +/// by id ... so file addressing and id addressing produce the same +/// realization by construction" (the driver's own doc comment). This pins +/// that property observably: registering a document and then running it once +/// by FILE and once by its own resulting content ID must refuse with the +/// byte-identical prose — proof the two addressing modes are not two +/// independently drifting code paths. No real data needed: both runs refuse +/// at the (data-free) v1-pipeline-shape preflight. +#[test] +fn campaign_run_by_content_id_matches_file_sugar_refusal() { + 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("mc2.process.json")), + ScratchPath::File(dir.join("mc2.campaign.json")), + ]); + let bp_id = seed_blueprint(dir, "campaign-run-id-seed"); + let proc_id = register_process_doc(dir, "mc2.process.json", MC_PROCESS_DOC); + write_doc(dir, "mc2.campaign.json", &campaign_doc_json(&bp_id, &proc_id, (1, 2), "", "")); + + let (file_out, file_code) = run_code_in(dir, &["campaign", "run", "mc2.campaign.json"]); + assert_eq!(file_code, Some(1), "stdout/stderr: {file_out}"); + + let (reg_out, reg_code) = run_code_in(dir, &["campaign", "register", "mc2.campaign.json"]); + assert_eq!(reg_code, Some(0), "register failed: {reg_out}"); + let id = reg_out + .lines() + .find(|l| l.starts_with("registered campaign content:")) + .expect("register line") + .trim_start_matches("registered campaign content:") + .split(' ') + .next() + .expect("id") + .to_string(); + let (id_out, id_code) = run_code_in(dir, &["campaign", "run", &id]); + assert_eq!(id_code, Some(1), "stdout/stderr: {id_out}"); + + let file_line = file_out.lines().find(|l| l.starts_with("aura: ")).expect("file refusal line"); + let id_line = id_out.lines().find(|l| l.starts_with("aura: ")).expect("id refusal line"); + assert_eq!(file_line, id_line, "file- and id-addressed runs must refuse identically"); +} + +/// `campaign run`'s file-sugar branch calls `parse_valid_campaign` directly +/// (unwrapped), the SAME intrinsic validator `campaign register` calls but +/// wraps in its own "refusing to register:" prefix. This pins that `run` +/// does NOT inherit register's wrapper — an intrinsically invalid document +/// refuses with the bare "campaign document invalid:" prose — and, like +/// register, never touches the store for it. No project scaffolding, no +/// data, no seeded blueprint needed: the refusal fires before any of that is +/// read. +#[test] +fn campaign_run_invalid_file_refuses_before_touching_store() { + // Run inside the built project (campaign run needs one, per the project + // gate); the document fails intrinsic validation before any store or + // referential check is reached, so the project's own store is untouched. + let _fixture = project_lock(); + let dir = built_project(); + let runs_dir = dir.join("runs"); + std::fs::remove_dir_all(&runs_dir).ok(); + let bad = CAMPAIGN_DOC.replacen(r#""values": [8]"#, r#""values": []"#, 1); + let bad_path = write_doc(dir, "bad.campaign.json", &bad); + let _cleanup = ScratchGuard(vec![ScratchPath::Dir(runs_dir.clone()), ScratchPath::File(bad_path)]); + + let (out, code) = run_code_in(dir, &["campaign", "run", "bad.campaign.json"]); + assert_eq!(code, Some(1), "stdout/stderr: {out}"); + assert!( + out.contains("aura: campaign document invalid:"), + "run must surface the bare doc-tier prose, not register's wrapper: {out}" + ); + assert!( + out.contains("axes.fast: an axis is a non-empty finite set"), + "the doc-tier fault names the offending axis: {out}" + ); + assert!(!out.contains("refusing to register:"), "run must not reuse register's prefix: {out}"); + assert!( + !dir.join("runs").join("campaigns").exists(), + "an invalid document must not create a store entry" + ); +}