Aura/crates/aura-cli/tests/project_sweep_campaign.rs

//! E2E (#235): the defining loop of a genuine external project — sweep MY OWN
//! project-authored node's knob — proven end to end.
//!
//! Every prior sweep/campaign in the suite ran over a std-only blueprint (or, at
//! most, a param-less project node); no project-authored node with an OPEN param
//! had ever gone through sweep or campaign. A fresh `aura new` project ships a
//! project-authored `Scale` node with one open `factor` param (#183); this test
//! authors an OPEN blueprint over that node (factor left unbound) and drives it
//! through the two acceptance boxes of #235:
//!
//!   1. `--list-axes` discovers the project node's own axis -> a real-data sweep
//!      over it -> family persisted -> `reproduce` re-derives N/N bit-identically;
//!   2. one campaign cell over the same project blueprint runs to exit 0.
//!
//! Sibling idiom: `project_new.rs` (scaffold via the built binary + cargo-build
//! the project crate in-test — the expensive-but-accepted authoring-loop idiom)
//! and `cli_run.rs`'s `local_data_present()` archive gate for the real-data legs.
//! The axis-discovery leg is NOT gated: it proves the project node's knob is
//! sweepable on every host, data or no data.

use std::path::{Path, PathBuf};
use std::process::{Command, Output};
use std::sync::{Mutex, MutexGuard, OnceLock};

const BIN: &str = env!("CARGO_BIN_EXE_aura");

// GER40 Sept-2024 window (inclusive Unix-ms) — the same gated real window the
// sibling sweep/campaign e2es drive (cli_run.rs / research_docs.rs).
const GER40_FROM_MS: &str = "1725148800000";
const GER40_TO_MS: &str = "1727740799999";

/// The OPEN blueprint: the scaffold's SMA-cross bias graph, but its terminal
/// node is the project's OWN `knob_lab::Scale` (named `gain`), left UNBOUND — so
/// `factor` is the one open sweep axis (`scaled_signal.gain.factor` wrapped /
/// `gain.factor` raw). Two-tier fixture (#241): `aura new knob-lab` scaffolds
/// the data-only project, and a sibling `knob-lab-nodes` node crate (attached
/// via `aura nodes new ... --namespace knob_lab`) supplies the `knob_lab`
/// vocabulary namespace — `Scale` is the starter node every node-crate
/// scaffold emits (#183).
const OPEN_BLUEPRINT: &str = r#"{"format_version":1,"blueprint":{"name":"scaled_signal","nodes":[{"primitive":{"type":"SMA","name":"fast","bound":[{"pos":0,"name":"length","kind":"I64","value":{"I64":2}}]}},{"primitive":{"type":"SMA","name":"slow","bound":[{"pos":0,"name":"length","kind":"I64","value":{"I64":4}}]}},{"primitive":{"type":"Sub"}},{"primitive":{"type":"Bias","name":"bias","bound":[{"pos":0,"name":"scale","kind":"F64","value":{"F64":0.5}}]}},{"primitive":{"type":"knob_lab::Scale","name":"gain"}}],"edges":[{"from":0,"to":2,"slot":0,"from_field":0},{"from":1,"to":2,"slot":1,"from_field":0},{"from":2,"to":3,"slot":0,"from_field":0},{"from":3,"to":4,"slot":0,"from_field":0}],"input_roles":[{"name":"price","targets":[{"node":0,"slot":0},{"node":1,"slot":0}],"source":"F64"}],"output":[{"node":4,"field":0,"name":"bias"}]}}"#;

/// The minimal executable process pipeline (one sweep stage) — the campaign
/// cell's methodology reference.
const SWEEP_ONLY_PROCESS: &str = r#"{"format_version":1,"kind":"process","name":"knob-sweep-only","pipeline":[{"block":"std::sweep","metric":"sqn_normalized","select":"argmax"}]}"#;

fn local_data_present() -> bool {
    Path::new(data_server::DEFAULT_DATA_PATH).is_dir()
}

fn aura_in(dir: &Path, args: &[&str]) -> Output {
    Command::new(BIN)
        .args(args)
        .current_dir(dir)
        .output()
        .expect("run aura")
}

/// Scaffold `knob-lab` once via the two-tier project path (#241): a data-only
/// `aura new` project, a sibling `knob-lab-nodes` node crate attached via
/// `aura nodes new` (with `--namespace knob_lab` overriding the crate-derived
/// default so the blueprint's `knob_lab::` prefix stays byte-identical), then
/// `cargo build` the node crate. Drops the OPEN blueprint into
/// `blueprints/scaled_open.json`. Returns the project dir. `OnceLock` so the
/// expensive scaffold+build happens once per test binary; the node crate's
/// path-dep resolves into this checkout (the baked engine root), like
/// `project_new.rs`. The base is a FIXED name under `CARGO_TARGET_TMPDIR`
/// (real disk, reclaimed by `cargo clean`), not a pid-suffixed tempdir: the
/// wipe below must hit the PREVIOUS run's copy, else every test-binary run
/// leaks the ~66 MB built scaffold — 159 of those filled the 16 GB /tmp
/// tmpfs to 100 % (#257).
fn built_scale_project() -> &'static PathBuf {
    static BUILT: OnceLock<PathBuf> = OnceLock::new();
    BUILT.get_or_init(|| {
        let base = Path::new(env!("CARGO_TARGET_TMPDIR")).join("aura-235-knob-lab");
        let _ = std::fs::remove_dir_all(&base);
        std::fs::create_dir_all(&base).expect("create scaffold base");

        let new = aura_in(&base, &["new", "knob-lab"]);
        assert!(
            new.status.success(),
            "aura new knob-lab failed: {}",
            String::from_utf8_lossy(&new.stderr)
        );
        let proj = base.join("knob-lab");

        let engine = env!("CARGO_MANIFEST_DIR").to_string() + "/../..";
        let nodes_new = aura_in(
            &proj,
            &[
                "nodes",
                "new",
                "knob-lab-nodes",
                "--namespace",
                "knob_lab",
                "--engine-path",
                &engine,
            ],
        );
        assert!(
            nodes_new.status.success(),
            "aura nodes new knob-lab-nodes failed: {}",
            String::from_utf8_lossy(&nodes_new.stderr)
        );
        let nodes_crate = base.join("knob-lab-nodes");

        let build = Command::new("cargo")
            .arg("build")
            .current_dir(&nodes_crate)
            .output()
            .expect("cargo build the scaffolded node crate");
        assert!(
            build.status.success(),
            "scaffolded node crate build failed:\n{}",
            String::from_utf8_lossy(&build.stderr)
        );

        std::fs::write(proj.join("blueprints/scaled_open.json"), OPEN_BLUEPRINT)
            .expect("write the open blueprint");
        proj
    })
}

/// Serializes tests that share the one built project (they reset `runs/`, so
/// parallel threads would race). Poison-tolerant.
fn project_lock() -> MutexGuard<'static, ()> {
    static LOCK: Mutex<()> = Mutex::new(());
    LOCK.lock().unwrap_or_else(|e| e.into_inner())
}

/// Removes the project's `runs/` on drop (including mid-panic) so a
/// content-addressed store never leaks between the shared tests.
struct RunsCleanup(PathBuf);
impl Drop for RunsCleanup {
    fn drop(&mut self) {
        std::fs::remove_dir_all(&self.0).ok();
    }
}

/// Enter the shared built project with a FRESH `runs/` store, serialized. Bind
/// the guard tuple for the whole test body: `let (dir, _g) = fresh_project();`.
/// `RunsCleanup` is ordered before the `MutexGuard` so the store removal runs
/// while the lock is still held.
fn fresh_project() -> (&'static PathBuf, (RunsCleanup, MutexGuard<'static, ()>)) {
    let lock = project_lock();
    let dir = built_scale_project();
    let runs = dir.join("runs");
    std::fs::remove_dir_all(&runs).ok();
    (dir, (RunsCleanup(runs), lock))
}

/// Acceptance box 1 (#235): a project-authored node with an OPEN param is
/// discoverable as a sweep axis, sweeps over real data into a persisted family,
/// and that family reproduces bit-identically — the defining research loop of a
/// genuine external project ("sweep my OWN node's knob"), proven end to end.
#[test]
fn project_node_open_param_sweeps_and_reproduces() {
    let (dir, _g) = fresh_project();

    // (a) `--list-axes` discovers the PROJECT node's own open knob, followed by
    // the fixture's three BOUND params as `default=`-lines (#246: every bound
    // param is an equally re-openable `--axis`, listed regardless of how many
    // knobs happen to be open alongside it). NOT gated: enumerating axes needs
    // no data, so the discoverability half runs on every host. The wrapped
    // name is `<blueprint>.<node>.<param>`.
    let axes = aura_in(dir, &["sweep", "blueprints/scaled_open.json", "--list-axes"]);
    assert!(
        axes.status.success(),
        "--list-axes stderr: {}",
        String::from_utf8_lossy(&axes.stderr)
    );
    assert_eq!(
        String::from_utf8_lossy(&axes.stdout),
        "scaled_signal.gain.factor:F64\n\
         scaled_signal.fast.length:I64 default=2\n\
         scaled_signal.slow.length:I64 default=4\n\
         scaled_signal.bias.scale:F64 default=0.5\n",
        "the project node's own OPEN param is the one open axis, alongside the \
         fixture's bound defaults; stderr: {}",
        String::from_utf8_lossy(&axes.stderr)
    );

    // The real-data sweep/reproduce legs are archive-gated.
    if !local_data_present() {
        eprintln!("skip: no local data at {}", data_server::DEFAULT_DATA_PATH);
        return;
    }

    // (b) sweep the project node's knob over real GER40 data -> two members.
    let sweep = aura_in(
        dir,
        &[
            "sweep",
            "blueprints/scaled_open.json",
            "--real",
            "GER40",
            "--from",
            GER40_FROM_MS,
            "--to",
            GER40_TO_MS,
            "--axis",
            "scaled_signal.gain.factor=0.5,1.0",
            "--name",
            "knob",
        ],
    );
    assert!(
        sweep.status.success(),
        "sweep stderr: {}",
        String::from_utf8_lossy(&sweep.stderr)
    );
    let stdout = String::from_utf8_lossy(&sweep.stdout).into_owned();
    let lines: Vec<&str> = stdout.lines().collect();
    assert_eq!(lines.len(), 2, "one member line per factor point: {stdout}");

    // Each member's manifest carries the swept PROJECT-node param binding under
    // its wrapped name — the load-bearing proof that it is MY node's knob that
    // varied across the family, not some incidental std axis.
    for (line, factor) in lines.iter().zip([0.5_f64, 1.0]) {
        let v: serde_json::Value = serde_json::from_str(line).expect("member line parses as JSON");
        assert_eq!(
            v["report"]["manifest"]["instrument"].as_str(),
            Some("GER40"),
            "member manifest stamps the instrument: {line}"
        );
        let params = v["report"]["manifest"]["params"]
            .as_array()
            .expect("manifest.params is an array");
        let bound = params
            .iter()
            .find(|p| p[0].as_str() == Some("scaled_signal.gain.factor"))
            .and_then(|p| p[1]["F64"].as_f64());
        assert_eq!(
            bound,
            Some(factor),
            "member manifest carries the swept project-node param binding: {line}"
        );
    }

    // (c) reproduce the persisted family: every member re-derives bit-identically.
    let family_id = serde_json::from_str::<serde_json::Value>(lines[0])
        .expect("first member line parses")["family_id"]
        .as_str()
        .expect("member line carries a family_id")
        .to_string();
    let rep = aura_in(dir, &["reproduce", &family_id]);
    assert!(
        rep.status.success(),
        "reproduce stderr: {}",
        String::from_utf8_lossy(&rep.stderr)
    );
    assert!(
        String::from_utf8_lossy(&rep.stdout).contains("reproduced 2/2 members bit-identically"),
        "reproduce stdout: {}",
        String::from_utf8_lossy(&rep.stdout)
    );
}

/// Acceptance box 2 (#235): one campaign cell over the same project blueprint.
/// The blueprint (with its project-authored open param) registers, a minimal
/// sweep-only process registers, and a campaign referencing both by content id
/// — with its axis on the project node's own `gain.factor` — runs to exit 0 with
/// a `campaign_run` record whose single cell links a sweep family.
#[test]
fn project_node_open_param_runs_one_campaign_cell() {
    let (dir, _g) = fresh_project();

    // Register the OPEN project blueprint -> content id.
    let reg = aura_in(dir, &["graph", "register", "blueprints/scaled_open.json"]);
    assert!(
        reg.status.success(),
        "graph register stderr: {}",
        String::from_utf8_lossy(&reg.stderr)
    );
    let reg_out = String::from_utf8_lossy(&reg.stdout).into_owned();
    let bp_id = reg_out
        .lines()
        .find(|l| l.starts_with("registered blueprint "))
        .expect("register line")
        .trim_start_matches("registered blueprint ")
        .split(' ')
        .next()
        .expect("blueprint id")
        .to_string();

    // Register a minimal sweep-only process -> content id.
    std::fs::write(dir.join("knob.process.json"), SWEEP_ONLY_PROCESS).expect("write process doc");
    let preg = aura_in(dir, &["process", "register", "knob.process.json"]);
    assert!(
        preg.status.success(),
        "process register stderr: {}",
        String::from_utf8_lossy(&preg.stderr)
    );
    let preg_out = String::from_utf8_lossy(&preg.stdout).into_owned();
    let proc_id = preg_out
        .lines()
        .find(|l| l.starts_with("registered process "))
        .expect("register line")
        .trim_start_matches("registered process ")
        .split(' ')
        .next()
        .expect("process id")
        .to_string();

    // Author the campaign cell: one instrument, one window, axis on the project
    // node's OWN `gain.factor` (the RAW param-space name campaign axes use).
    let campaign = format!(
        r#"{{
  "format_version": 1,
  "kind": "campaign",
  "name": "knob-cell",
  "data": {{ "instruments": ["GER40"], "windows": [ {{ "from_ms": {from}, "to_ms": {to} }} ] }},
  "strategies": [ {{ "ref": {{ "content_id": "{bp}" }},
                    "axes": {{ "gain.factor": {{ "kind": "F64", "values": [0.5, 1.0] }} }} }} ],
  "process": {{ "ref": {{ "content_id": "{proc}" }} }},
  "seed": 7,
  "presentation": {{ "persist_taps": [], "emit": ["family_table"] }}
}}"#,
        from = GER40_FROM_MS,
        to = GER40_TO_MS,
        bp = bp_id,
        proc = proc_id,
    );
    std::fs::write(dir.join("knob.campaign.json"), &campaign).expect("write campaign doc");

    let run = aura_in(dir, &["campaign", "run", "knob.campaign.json"]);
    let out = format!(
        "{}{}",
        String::from_utf8_lossy(&run.stdout),
        String::from_utf8_lossy(&run.stderr)
    );

    // Archive-gated: skip cleanly on a data-less machine (the member-data
    // refusal, never a panic) — mirroring campaign_run_real_e2e_*.
    if run.status.code() == Some(1)
        && (out.contains("no recorded geometry") || out.contains("no data for instrument"))
    {
        eprintln!("skip: no local GER40 data for the campaign cell");
        return;
    }

    assert_eq!(run.status.code(), Some(0), "campaign run: {out}");
    let record = 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).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 over the project blueprint: {record}"
    );
    assert!(
        cells[0]["stages"][0]["family_id"].as_str().is_some(),
        "the sweep stage links a family: {record}"
    );
}