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

//! #282: a hand-authored blueprint with a declared tap, run on the single-run
//! path, records the tapped series; the same blueprint in a sweep does not
//! (`run_blueprint_member`, the sweep/reduce path, never calls `bind_tap` —
//! structurally guaranteed by Task 6 leaving it untouched, not re-asserted
//! here).

use std::path::Path;
use std::process::Command;

/// Path to the freshly-built `aura` binary (Cargo sets this env var for the
/// test crate; the binary is named `aura` in `Cargo.toml`).
const BIN: &str = env!("CARGO_BIN_EXE_aura");

/// A fresh, unique working directory for this process test (so the run's
/// `./runs/traces/` never dirties the repo). Unique per test + per process.
fn temp_cwd(name: &str) -> std::path::PathBuf {
    let dir = Path::new(env!("CARGO_TARGET_TMPDIR")).join(format!("aura-cli-tap-{name}"));
    let _ = std::fs::remove_dir_all(&dir);
    std::fs::create_dir_all(&dir).expect("create temp cwd");
    dir
}

/// The exact synthetic price array `r_sma_prices()` (main.rs) feeds the
/// `RunData::Synthetic` path — duplicated here (not re-exported; a private
/// CLI helper) so the tapped SMA(2) column can be pinned against an
/// independently-derived expectation, not a blind capture.
const R_SMA_PRICES: [f64; 18] = [
    1.0000, 1.0008, 1.0021, 1.0039, 1.0062, 1.0090, 1.0083, 1.0061, 1.0034, 1.0012, 0.9998,
    1.0006, 1.0024, 1.0047, 1.0069, 1.0086, 1.0097, 1.0092,
];

/// The shipped `examples/r_sma.json` blueprint, with one additional declared
/// tap on node 0 ("fast", `SMA(length=2)`) field 0 — the smallest addition
/// that names an interior producer output without touching the existing
/// topology (nodes/edges/input_roles/output all verbatim).
fn tap_blueprint_json() -> String {
    let path = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let doc = std::fs::read_to_string(path).expect("read examples/r_sma.json");
    let mut v: serde_json::Value = serde_json::from_str(&doc).expect("parse r_sma.json");
    v["blueprint"]["taps"] = serde_json::json!([{"name": "fast_tap", "from": {"node": 0, "field": 0}}]);
    serde_json::to_string(&v).expect("re-serialize tapped blueprint")
}

/// Property: on the single-run path (`aura run <blueprint.json>` over the
/// built-in synthetic stream, no `--real`), a declared tap is bound and its
/// per-cycle series is persisted through the trace store — the persisted
/// `fast_tap.json` carries exactly the SMA(2) values over the known synthetic
/// price stream (silent until warm at sample 2, one value per cycle after).
#[test]
fn single_run_persists_a_declared_tap_series() {
    let cwd = temp_cwd("single-run-persists");
    let bp_path = cwd.join("tapped_r_sma.json");
    std::fs::write(&bp_path, tap_blueprint_json()).expect("write tapped blueprint");

    let out = Command::new(BIN)
        .args(["run", bp_path.to_str().expect("utf-8 path")])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura run");
    assert!(out.status.success(), "stderr: {}", String::from_utf8_lossy(&out.stderr));

    let trace_path = cwd.join("runs/traces/sma_signal/fast_tap.json");
    let trace_text = std::fs::read_to_string(&trace_path).unwrap_or_else(|e| {
        panic!("expected a persisted tap trace at {}: {e}", trace_path.display())
    });
    let trace: serde_json::Value = serde_json::from_str(&trace_text).expect("parse tap trace json");

    assert_eq!(trace["tap"], "fast_tap");
    assert_eq!(trace["kinds"], serde_json::json!(["F64"]));

    // SMA(2), silent until warm (2 samples): first emission at the 2nd sample
    // (ts=2), value = mean(prices[0], prices[1]); then a rolling mean of the
    // last two prices per subsequent cycle (ts = i+1 for prices[i]).
    let expected: Vec<f64> = (1..R_SMA_PRICES.len())
        .map(|i| (R_SMA_PRICES[i - 1] + R_SMA_PRICES[i]) / 2.0)
        .collect();
    let expected_ts: Vec<i64> = (2..=R_SMA_PRICES.len() as i64).collect();

    let got_ts: Vec<i64> = trace["ts"].as_array().expect("ts array").iter().map(|v| v.as_i64().unwrap()).collect();
    assert_eq!(got_ts, expected_ts, "recorded timestamps");

    let got: Vec<f64> = trace["columns"][0]
        .as_array()
        .expect("columns[0] array")
        .iter()
        .map(|v| v.as_f64().unwrap())
        .collect();
    assert_eq!(got.len(), expected.len(), "recorded row count");
    for (i, (g, e)) in got.iter().zip(expected.iter()).enumerate() {
        assert!((g - e).abs() < 1e-9, "row {i}: got {g}, expected {e}");
    }
}

/// A tap-free blueprint run (the existing `run_prints_json_and_exits_zero`
/// shape) leaves no `runs/` directory at all — the tap machinery is fully
/// inert (no binds, no trace_store write) when a blueprint declares no taps,
/// matching Task 6's byte-identical guarantee.
#[test]
fn tap_free_run_writes_no_trace_store() {
    let cwd = temp_cwd("tap-free-writes-nothing");
    let bp = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let out = Command::new(BIN)
        .args(["run", &bp])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura run");
    assert!(out.status.success(), "stderr: {}", String::from_utf8_lossy(&out.stderr));
    assert!(!cwd.join("runs").exists(), "a tap-free run must not create a trace store");
}

/// The shipped `examples/r_sma.json` blueprint with TWO declared taps that
/// share the same name (`"dup"`), on two distinct producers (node 0 "fast" and
/// node 1 "slow") — the shape `run_signal_r`'s `seen: BTreeSet` dedup guard
/// exists for (`FlatGraph::bind_tap` itself keeps no cross-call state per its
/// own doc comment, so nothing upstream of the CLI catches this).
fn duplicate_tap_blueprint_json() -> String {
    let path = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let doc = std::fs::read_to_string(path).expect("read examples/r_sma.json");
    let mut v: serde_json::Value = serde_json::from_str(&doc).expect("parse r_sma.json");
    v["blueprint"]["taps"] = serde_json::json!([
        {"name": "dup", "from": {"node": 0, "field": 0}},
        {"name": "dup", "from": {"node": 1, "field": 0}},
    ]);
    serde_json::to_string(&v).expect("re-serialize duplicate-tap blueprint")
}

/// The shipped `examples/r_sma.json` blueprint with two distinctly named
/// declared taps: `fast_tap` on node 0 ("fast", SMA(2)) and `slow_tap` on
/// node 1 ("slow") — the smallest fixture on which an explicit `--tap`
/// plan and the record-all default can be compared file-by-file (#310).
fn two_tap_blueprint_json() -> String {
    let path = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let doc = std::fs::read_to_string(path).expect("read examples/r_sma.json");
    let mut v: serde_json::Value = serde_json::from_str(&doc).expect("parse r_sma.json");
    v["blueprint"]["taps"] = serde_json::json!([
        {"name": "fast_tap", "from": {"node": 0, "field": 0}},
        {"name": "slow_tap", "from": {"node": 1, "field": 0}},
    ]);
    serde_json::to_string(&v).expect("re-serialize two-tap blueprint")
}

/// Property: **a blueprint declaring two taps under the same name is refused
/// on the single-run path with a named error and exit code 1, before any
/// trace is persisted** — the CLI-owned `TapBindError::DuplicateBind` dedup
/// guard (bind_tap itself binds a single tap and keeps no cross-call state,
/// per its own doc comment) actually fires end-to-end, and the partial first
/// bind never reaches disk as a half-written trace store.
#[test]
fn single_run_refuses_a_duplicate_tap_name_before_persisting_anything() {
    let cwd = temp_cwd("duplicate-tap-name-refused");
    let bp_path = cwd.join("dup_tap_r_sma.json");
    std::fs::write(&bp_path, duplicate_tap_blueprint_json()).expect("write dup-tap blueprint");

    let out = Command::new(BIN)
        .args(["run", bp_path.to_str().expect("utf-8 path")])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura run");

    assert!(!out.status.success(), "a duplicate tap name must be refused, not silently bound");
    let stderr = String::from_utf8_lossy(&out.stderr);
    assert!(
        stderr.contains("dup") && stderr.contains("more than once"),
        "stderr must name the offending tap and the duplicate-bind fault: {stderr}"
    );
    assert!(!cwd.join("runs").exists(), "a refused run must not persist a half-written trace store");
}

/// The `aura: writing tap traces failed: …` register, pre-run arm: `runs`
/// exists as a FILE, so the trace store's directory creation
/// (`begin_run`) fails before the run — exit 1 through the register.
/// (This register was code-present but untested before #283.)
#[test]
fn an_unwritable_store_root_exits_through_the_tap_trace_register() {
    let cwd = temp_cwd("store-root-is-a-file");
    let bp_path = cwd.join("tapped_r_sma.json");
    std::fs::write(&bp_path, tap_blueprint_json()).expect("write tapped blueprint");
    std::fs::write(cwd.join("runs"), b"not a directory").expect("occupy runs as a file");

    let out = Command::new(BIN)
        .args(["run", bp_path.to_str().expect("utf-8 path")])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura run");

    assert!(!out.status.success(), "an unwritable store must refuse, not succeed");
    let stderr = String::from_utf8_lossy(&out.stderr);
    assert!(
        stderr.contains("writing tap traces failed"),
        "the tap-trace register must fire: {stderr}"
    );
}

/// The same register, deferred arm (#283): the run directory pre-exists
/// read-only, so `begin_run` succeeds (create_dir_all on an existing dir)
/// but the record consumer's deferred open in `initialize` fails; the run
/// COMPLETES, the failure surfaces terminally through the register, and no
/// `index.json` is written (the store's treat-as-absent crash shape).
#[cfg(unix)]
#[test]
fn a_read_only_run_dir_surfaces_terminally_through_the_tap_trace_register() {
    use std::os::unix::fs::PermissionsExt;

    let cwd = temp_cwd("run-dir-read-only");
    let bp_path = cwd.join("tapped_r_sma.json");
    std::fs::write(&bp_path, tap_blueprint_json()).expect("write tapped blueprint");
    let run_dir = cwd.join("runs/traces/sma_signal");
    std::fs::create_dir_all(&run_dir).expect("pre-create run dir");
    std::fs::set_permissions(&run_dir, std::fs::Permissions::from_mode(0o555))
        .expect("make run dir read-only");

    let out = Command::new(BIN)
        .args(["run", bp_path.to_str().expect("utf-8 path")])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura run");

    // restore permissions FIRST so the next run's temp_cwd cleanup works.
    std::fs::set_permissions(&run_dir, std::fs::Permissions::from_mode(0o755))
        .expect("restore run dir permissions");

    assert!(!out.status.success(), "a failed writer must surface, not pass silently");
    let stderr = String::from_utf8_lossy(&out.stderr);
    assert!(
        stderr.contains("writing tap traces failed"),
        "the tap-trace register must fire terminally: {stderr}"
    );
    assert!(!run_dir.join("index.json").exists(), "no index — treat-as-absent crash shape");
}

/// #310: `--tap fast_tap=mean` subscribes the declared tap to the `mean`
/// fold instead of the record-all default — the trace store holds ONE
/// summary row whose value is the mean of the full SMA(2) series, and
/// `index.json` lists exactly the subscribed tap.
#[test]
fn run_tap_selector_persists_a_fold_summary_row() {
    let cwd = temp_cwd("tap-selector-mean");
    let bp_path = cwd.join("tapped_r_sma.json");
    std::fs::write(&bp_path, tap_blueprint_json()).expect("write tapped blueprint");

    let out = Command::new(BIN)
        .args(["run", bp_path.to_str().expect("utf-8 path"), "--tap", "fast_tap=mean"])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura run");
    assert!(out.status.success(), "stderr: {}", String::from_utf8_lossy(&out.stderr));

    let index_text = std::fs::read_to_string(cwd.join("runs/traces/sma_signal/index.json"))
        .expect("read index.json");
    let index: serde_json::Value = serde_json::from_str(&index_text).expect("parse index.json");
    assert_eq!(index["taps"], serde_json::json!(["fast_tap"]));

    let trace_text = std::fs::read_to_string(cwd.join("runs/traces/sma_signal/fast_tap.json"))
        .expect("read fold trace");
    let trace: serde_json::Value = serde_json::from_str(&trace_text).expect("parse fold trace");
    assert_eq!(trace["tap"], "fast_tap");
    let rows = trace["columns"][0].as_array().expect("columns[0] array");
    assert_eq!(rows.len(), 1, "a fold lands exactly one summary row");
    let sma: Vec<f64> = (1..R_SMA_PRICES.len())
        .map(|i| (R_SMA_PRICES[i - 1] + R_SMA_PRICES[i]) / 2.0)
        .collect();
    let expected = sma.iter().sum::<f64>() / sma.len() as f64;
    let got = rows[0].as_f64().expect("f64 row");
    assert!((got - expected).abs() < 1e-9, "mean row: got {got}, expected {expected}");
}

/// #310: an all-`record` explicit plan is byte-identical to the no-flag
/// record-all default — the selector replaces the default without
/// changing what `record` itself writes (C1 determinism across runs).
#[test]
fn run_explicit_record_plan_matches_the_record_all_default() {
    let cwd_a = temp_cwd("record-default");
    let cwd_b = temp_cwd("record-explicit");
    for cwd in [&cwd_a, &cwd_b] {
        std::fs::write(cwd.join("two_tap.json"), two_tap_blueprint_json())
            .expect("write two-tap blueprint");
    }
    let run = |cwd: &std::path::Path, extra: &[&str]| {
        let mut args = vec!["run", "two_tap.json"];
        args.extend_from_slice(extra);
        let out = Command::new(BIN)
            .args(&args)
            .current_dir(cwd)
            .output()
            .expect("spawn aura run");
        assert!(out.status.success(), "stderr: {}", String::from_utf8_lossy(&out.stderr));
    };
    run(&cwd_a, &[]);
    run(&cwd_b, &["--tap", "fast_tap=record", "--tap", "slow_tap=record"]);

    for file in ["fast_tap.json", "slow_tap.json"] {
        let a = std::fs::read(cwd_a.join("runs/traces/sma_signal").join(file)).expect("default trace");
        let b = std::fs::read(cwd_b.join("runs/traces/sma_signal").join(file)).expect("explicit trace");
        assert_eq!(a, b, "{file} must be byte-identical across default and explicit record plans");
    }
    let taps_of = |cwd: &std::path::Path| -> serde_json::Value {
        let text = std::fs::read_to_string(cwd.join("runs/traces/sma_signal/index.json"))
            .expect("read index.json");
        serde_json::from_str::<serde_json::Value>(&text).expect("parse index.json")["taps"].clone()
    };
    assert_eq!(taps_of(&cwd_a), taps_of(&cwd_b));
}

/// #310: an unknown fold label is refused through the registry's
/// roster-enumerating refusal, before any store write.
#[test]
fn run_tap_selector_refuses_an_unknown_label_with_the_roster() {
    let cwd = temp_cwd("tap-selector-unknown-label");
    let bp_path = cwd.join("tapped_r_sma.json");
    std::fs::write(&bp_path, tap_blueprint_json()).expect("write tapped blueprint");
    let out = Command::new(BIN)
        .args(["run", bp_path.to_str().expect("utf-8 path"), "--tap", "fast_tap=medain"])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura run");
    assert!(!out.status.success(), "an unknown fold label must refuse");
    let stderr = String::from_utf8_lossy(&out.stderr);
    assert!(
        stderr.contains("medain") && stderr.contains("mean"),
        "refusal must name the label and enumerate the roster: {stderr}"
    );
    assert!(!cwd.join("runs").exists(), "a refused run must not write the store");
}

/// #310: a selection naming a tap the blueprint does not declare is
/// refused typed (UndeclaredTap), before any store write.
#[test]
fn run_tap_selector_refuses_an_undeclared_tap() {
    let cwd = temp_cwd("tap-selector-undeclared");
    let bp_path = cwd.join("tapped_r_sma.json");
    std::fs::write(&bp_path, tap_blueprint_json()).expect("write tapped blueprint");
    let out = Command::new(BIN)
        .args(["run", bp_path.to_str().expect("utf-8 path"), "--tap", "nope=mean"])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura run");
    assert!(!out.status.success(), "an undeclared tap must refuse");
    let stderr = String::from_utf8_lossy(&out.stderr);
    assert!(stderr.contains("nope"), "refusal must name the offending tap: {stderr}");
    assert!(!cwd.join("runs").exists(), "a refused run must not write the store");
}

/// #310: a malformed or duplicate `--tap` is a usage error (exit 2)
/// before anything runs.
#[test]
fn run_tap_selector_refuses_malformed_and_duplicate_pairs() {
    let cwd = temp_cwd("tap-selector-usage");
    let bp_path = cwd.join("tapped_r_sma.json");
    std::fs::write(&bp_path, tap_blueprint_json()).expect("write tapped blueprint");
    for args in [
        vec!["--tap", "fast_tapmean"],
        vec!["--tap", "fast_tap=mean", "--tap", "fast_tap=last"],
    ] {
        let mut full = vec!["run", bp_path.to_str().expect("utf-8 path")];
        full.extend(args);
        let out = Command::new(BIN)
            .args(&full)
            .current_dir(&cwd)
            .output()
            .expect("spawn aura run");
        assert_eq!(out.status.code(), Some(2), "usage errors exit 2");
        let stderr = String::from_utf8_lossy(&out.stderr);
        assert!(stderr.contains("--tap"), "usage message names the flag: {stderr}");
        assert!(!cwd.join("runs").exists());
    }
}