//! Integration test: drive the built `aura` binary as a downstream user would,
//! asserting the `run` subcommand's stdout/exit contract and the bad-args path.

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

mod common;
use common::{fresh_project, fresh_project_with_data, fresh_project_with_gapped_data};

/// 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 a process test that persists run
/// records (so `aura sweep`'s `./runs/runs.jsonl` never dirties the repo).
/// Unique per test + per process; no external tempfile dependency.
fn temp_cwd(name: &str) -> std::path::PathBuf {
    let dir = Path::new(env!("CARGO_TARGET_TMPDIR")).join(format!("aura-cli-{name}"));
    let _ = std::fs::remove_dir_all(&dir);
    std::fs::create_dir_all(&dir).expect("create temp cwd");
    dir
}

/// #218: a well-formed dissolved verb run outside a project refuses up front
/// (exit 1, "needs a project") and leaves no store — parity with `campaign run`.
/// The refusal names the INVOKED verb, not a copy-pasted generic string — a
/// per-verb literal shared across four call sites (`validate_and_register_axes`
/// takes `verb` as a parameter) is exactly the kind of edit that silently
/// regresses to one hardcoded verb name; this pins that each call site still
/// threads its own name through.
#[test]
fn sweep_outside_project_refuses_and_leaves_no_store() {
    let cwd = temp_cwd("sweep-outside-project-refuses");
    let bp = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args(["sweep", &bp, "--real", "GER40", "--axis", "sma_signal.fast.length=2,4"])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura");
    assert_eq!(out.status.code(), Some(1), "stderr: {}", String::from_utf8_lossy(&out.stderr));
    assert!(String::from_utf8_lossy(&out.stderr).contains("sweep needs a project"));
    assert!(!cwd.join("runs").exists(), "a refused run must leave no store");
}

/// #218: same refusal as `sweep_outside_project_refuses_and_leaves_no_store`,
/// for `generalize`'s instrument-list form — the message names `generalize`,
/// not `sweep` (see that test's doc comment for the property).
#[test]
fn generalize_outside_project_refuses_and_leaves_no_store() {
    let cwd = temp_cwd("generalize-outside-project-refuses");
    let bp = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args(["generalize", &bp, "--real", "GER40,USDJPY", "--axis", "sma_signal.fast.length=2"])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura");
    assert_eq!(out.status.code(), Some(1), "stderr: {}", String::from_utf8_lossy(&out.stderr));
    assert!(String::from_utf8_lossy(&out.stderr).contains("generalize needs a project"));
    assert!(!cwd.join("runs").exists(), "a refused run must leave no store");
}

/// #218: same refusal as `sweep_outside_project_refuses_and_leaves_no_store`,
/// for `walkforward`'s valid `--axis` form — the message names `walkforward`,
/// not `sweep` (see that test's doc comment for the property).
#[test]
fn walkforward_outside_project_refuses_and_leaves_no_store() {
    let cwd = temp_cwd("walkforward-outside-project-refuses");
    let bp = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args(["walkforward", &bp, "--real", "GER40", "--axis", "sma_signal.fast.length=2,4"])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura");
    assert_eq!(out.status.code(), Some(1), "stderr: {}", String::from_utf8_lossy(&out.stderr));
    assert!(String::from_utf8_lossy(&out.stderr).contains("walkforward needs a project"));
    assert!(!cwd.join("runs").exists(), "a refused run must leave no store");
}

/// #218: same refusal as `sweep_outside_project_refuses_and_leaves_no_store`,
/// for `mc`'s valid `--axis` form — the message names `mc`, not `sweep` (see
/// that test's doc comment for the property).
#[test]
fn mc_outside_project_refuses_and_leaves_no_store() {
    let cwd = temp_cwd("mc-outside-project-refuses");
    let bp = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args(["mc", &bp, "--real", "GER40", "--axis", "sma_signal.fast.length=2,4"])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura");
    assert_eq!(out.status.code(), Some(1), "stderr: {}", String::from_utf8_lossy(&out.stderr));
    assert!(String::from_utf8_lossy(&out.stderr).contains("mc needs a project"));
    assert!(!cwd.join("runs").exists(), "a refused run must leave no store");
}

/// Property (#275, retrofitted): `run_signal_r` now keys its opened sources by
/// role name (`h.run_bound(key_supply(&binding, sources))`) instead of feeding
/// the harness positionally (`h.run(sources)`). This pre-existing happy-path
/// pin therefore doubles as the by-name-binding transparency check for that one
/// call site: a well-bound strategy's by-name path resolves back to the exact
/// same source order `bind_sources` was given, so the migration is invisible
/// here — unchanged exit code, unchanged JSON output byte-for-byte. It does NOT
/// exercise `run_blueprint_member` or the campaign re-run/trace path — those
/// two migrated call sites are covered by the sweep/campaign e2e fixtures
/// elsewhere in this crate, not by this test.
#[test]
fn run_prints_json_and_exits_zero() {
    // `run` is blueprint-only now (#159 cut 4 retired the bare built-in default);
    // the shipped r-sma example is the deterministic reference blueprint.
    let out = Command::new(BIN).args(["run", "examples/r_sma.json"]).output().expect("spawn aura run");
    assert!(out.status.success(), "exit status: {:?}", out.status);

    let stdout = String::from_utf8(out.stdout).expect("utf-8 stdout");
    // exactly one line (the JSON object + a trailing newline from println!).
    assert_eq!(stdout.lines().count(), 1, "stdout was: {stdout:?}");
    let line = stdout.trim_end();

    // canonical cycle-0009 JSON shape: nested manifest + metrics, stable keys.
    // The manifest leads with a `commit` field; its value is the build's git
    // identity (asserted by `run_manifest_commit_carries_real_git_head`), so
    // this shape check pins only the surrounding structure, not the value.
    assert!(line.starts_with("{\"manifest\":{\"commit\":\""), "got: {line}");
    assert!(line.contains("\"broker\":\"sim-optimal+risk-executor(pip_size=0.0001)\""), "got: {line}");
    assert!(line.contains("\"window\":[1,18]"), "got: {line}");
    assert!(line.contains("\"metrics\":{\"total_pips\":"), "got: {line}");
    // the r-sma harness carries an R block (unlike the retired pip-only default).
    assert!(line.contains("\"r\":{\"expectancy_r\":"), "got: {line}");
}

/// Property (#249): a plain `aura run` on a fully bound blueprint records the
/// untouched bound param values *directly in the outcome artifact*, so a raw
/// reader of the manifest recovers the effective params without dereferencing
/// the topology hash into the content-addressed store and re-implementing
/// aura's merge semantics. The manifest carries a `defaults` field — the bound
/// params merged inside `PrimitiveBuilder::build` below every record-writing
/// layer — as wrap-prefixed `(name, Scalar)` pairs in `bound_param_space()`
/// order (the same coordinate `--axis` uses). `params` keeps its "what varied"
/// meaning: on a plain run nothing varied, so it stays `[]`, and every bound
/// value lands in `defaults` instead of vanishing from the record.
#[test]
fn run_manifest_stamps_untouched_bound_defaults() {
    // The shipped r_sma.json is fully bound (fast.length=2, slow.length=4,
    // bias.scale=0.5); no axis re-opens any of them, so none flow through the
    // param space — today they appear in no outcome artifact ("params":[]).
    let out = Command::new(BIN).args(["run", "examples/r_sma.json"]).output().expect("spawn aura run");
    assert!(out.status.success(), "exit status: {:?}", out.status);
    let stdout = String::from_utf8(out.stdout).expect("utf-8 stdout");
    let line = stdout.trim_end();

    // "what varied" is empty on a plain run — the bound defaults are NOT params.
    assert!(line.contains("\"params\":[]"), "params stays empty on a plain run: {line}");

    // The untouched bound values are stamped in `defaults`, wrap-prefixed with
    // the blueprint name and ordered by `bound_param_space()` (node-declaration
    // order: fast, slow, bias), each a self-describing tagged Scalar (C14).
    assert!(
        line.contains(
            r#""defaults":[["sma_signal.fast.length",{"I64":2}],["sma_signal.slow.length",{"I64":4}],["sma_signal.bias.scale",{"F64":0.5}]]"#
        ),
        "manifest must carry the untouched bound defaults: {line}",
    );
}

/// Property: the RunManifest is self-identifying — its `commit` carries the
/// real code identity that produced the run, not a placeholder. C8/C18 audit
/// trail (this run = this commit): once runs are archived, `manifest.commit`
/// must point back at the source. The build captures the current git HEAD at
/// compile time; the `"unknown"` literal survives only when there is no git.
///
/// Structural assertion (not exact-equality) on purpose: the working tree may
/// be dirty when this runs (e.g. this very test file uncommitted), so a build
/// that appends a `-dirty` suffix would not equal `rev-parse HEAD` verbatim.
/// The honest contract is: the field CONTAINS the current HEAD sha AND is not
/// the bare `"unknown"` placeholder — true regardless of dirtiness/suffix.
#[test]
fn run_manifest_commit_carries_real_git_head() {
    // The current HEAD sha, obtained the same way the build is expected to.
    let head = Command::new("git")
        .args(["rev-parse", "HEAD"])
        .output()
        .expect("spawn git rev-parse HEAD");
    assert!(head.status.success(), "git rev-parse HEAD failed");
    let head_sha = String::from_utf8(head.stdout).expect("utf-8 sha");
    let head_sha = head_sha.trim();
    assert!(!head_sha.is_empty(), "empty HEAD sha");

    // Drive the binary and pull `manifest.commit` out of the single JSON line.
    // `run` is blueprint-only now (#159 cut 4); the shipped r-sma example stands in.
    let out = Command::new(BIN).args(["run", "examples/r_sma.json"]).output().expect("spawn aura run");
    assert!(out.status.success(), "exit status: {:?}", out.status);
    let stdout = String::from_utf8(out.stdout).expect("utf-8 stdout");
    let line = stdout.trim_end();

    // Locate the `"commit":"<value>"` value without a JSON dependency (the
    // sibling tests parse this output by substring too).
    let key = "\"commit\":\"";
    let start = line.find(key).expect("manifest has a commit field") + key.len();
    let rest = &line[start..];
    let end = rest.find('"').expect("commit field is a closed string");
    let commit = &rest[..end];

    assert_ne!(
        commit, "unknown",
        "manifest.commit is still the placeholder: {line}"
    );
    assert!(
        commit.contains(head_sha),
        "manifest.commit {commit:?} should contain the current HEAD sha {head_sha:?}; line: {line}"
    );
}

/// Property: `aura run` is strict about its argument vector — a trailing
/// unexpected token (e.g. a typo'd flag like `aura run --sweep`) takes the
/// error path (usage on stderr, exit 2), never a silent full run. This is the
/// strict reading ratified in #16: keying only on the first token being `run`
/// and ignoring the rest would let a typo masquerade as a successful run.
/// The same test pins positive-preservation — a valid `aura run <blueprint.json>`
/// still emits the JSON report on stdout and exits 0 — so the strict guard cannot
/// regress the happy path. (`run` is blueprint-only now, #159 cut 4: there is no
/// more bare/built-in default to preserve, so the positive case uses the shipped
/// r-sma example.)
#[test]
fn run_with_trailing_token_is_strict_and_exits_two() {
    // Negative: an unexpected trailing token is rejected like a bad-args call.
    let out = Command::new(BIN)
        .args(["run", "extra-garbage"])
        .output()
        .expect("spawn aura run extra-garbage");
    assert_eq!(
        out.status.code(),
        Some(2),
        "`aura run extra-garbage` must reject the trailing token: {:?}",
        out.status
    );
    assert!(
        out.stdout.is_empty(),
        "stdout should be empty on the usage path, got: {:?}",
        String::from_utf8_lossy(&out.stdout)
    );
    let stderr = String::from_utf8(out.stderr).expect("utf-8 stderr");
    assert!(stderr.contains("Usage"), "stderr was: {stderr:?}");

    // Positive-preservation: a valid blueprint invocation still runs and prints the report.
    let ok = Command::new(BIN)
        .args(["run", "examples/r_sma.json"])
        .output()
        .expect("spawn aura run examples/r_sma.json");
    assert_eq!(
        ok.status.code(),
        Some(0),
        "a valid `aura run <blueprint.json>` must still succeed: {:?}",
        ok.status
    );
    let ok_stdout = String::from_utf8(ok.stdout).expect("utf-8 stdout");
    assert!(
        ok_stdout.trim_start().starts_with("{\"manifest\":"),
        "a valid run should print the JSON report, got: {ok_stdout:?}"
    );
}

#[test]
fn run_real_no_geometry_symbol_refuses_with_exit_1() {
    // The geometry-sidecar lookup precedes any bar-data access, so a symbol with no
    // recorded geometry refuses with NO local data required (CI-safe). "NONEXISTENT"
    // has no recorded geometry on any host. `run` is blueprint-only now (#159 cut 4);
    // the shipped r-sma example reaches the same pip-refusal gate in `run_data_from`.
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args(["run", "examples/r_sma.json", "--real", "NONEXISTENT"])
        .output()
        .expect("spawn aura");
    assert_eq!(out.status.code(), Some(1), "expected exit 1");
    let stderr = String::from_utf8_lossy(&out.stderr);
    assert!(
        stderr.contains("no recorded geometry for symbol 'NONEXISTENT'"),
        "expected the per-instrument-pip refusal message, got: {stderr}"
    );
}

/// Property: the pip-refusal is sourced ONLY from recorded provider geometry — the
/// removed authored instrument floor (the hand-authored "vetted" table) no
/// longer participates in the refusal path. The cycle-0074 headline is the floor's
/// REMOVAL, not a rename; the observable signature is that the refusal message
/// never points the user at a hand-authored table to extend. A regression that
/// re-introduced an authored fallback (or restored the old "add it to the
/// instrument table" wording) would re-leak this vocabulary. Asserts the OLD
/// authored-floor phrasing is ABSENT — the negative complement to the positive
/// "no recorded geometry" assertion above. Archive-independent: "NONEXISTENT" has
/// no sidecar on any host, so the refusal fires without local data.
#[test]
fn run_real_refusal_names_no_authored_floor() {
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args(["run", "examples/r_sma.json", "--real", "NONEXISTENT"])
        .output()
        .expect("spawn aura");
    assert_eq!(out.status.code(), Some(1), "expected exit 1");
    let stderr = String::from_utf8_lossy(&out.stderr);
    assert!(
        !stderr.contains("instrument table"),
        "refusal must not point at a removed authored table, got: {stderr}"
    );
    assert!(
        !stderr.contains("vetted"),
        "refusal must not use the removed vetted-floor vocabulary, got: {stderr}"
    );
    assert!(
        !stderr.contains("instrument spec"),
        "refusal must not reference the removed authored instrument floor, got: {stderr}"
    );
}

/// Property: the un-specced-symbol refusal is a CLEAN refusal — it emits NOTHING
/// on stdout, never a partial `RunReport`. The #22 acceptance is "refuse INSTEAD
/// of emitting nonsense": the pip lookup precedes harness construction and data
/// access, so a `None` spec must short-circuit to `exit(1)` before any JSON line
/// can reach stdout. A regression that moved the lookup after a partial run would
/// leak a `{"manifest":...}` line here while still exiting non-zero; this pins
/// that it cannot. Archive-independent (the lookup never touches local data).
#[test]
fn run_real_no_geometry_symbol_emits_no_stdout_report() {
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args(["run", "examples/r_sma.json", "--real", "NONEXISTENT"])
        .output()
        .expect("spawn aura");
    assert_eq!(out.status.code(), Some(1), "expected exit 1: {:?}", out.status);
    assert!(
        out.stdout.is_empty(),
        "the refusal must not leak a partial report to stdout, got: {:?}",
        String::from_utf8_lossy(&out.stdout)
    );
}

/// Property: the pip-honesty refusal keys on the SYMBOL having no recorded geometry,
/// not on `--real` being taken at all. A symbol *with* a recorded sidecar (`EURUSD`)
/// gets PAST the geometry lookup, so it never produces the pip-refusal message — it
/// either runs to a report (local data present) or hits the DISTINCT no-local-data
/// usage path ("no local data for symbol"). Either way the "no recorded geometry"
/// string is absent. This distinguishes the per-instrument-pip honesty lever from
/// the unrelated arg-grammar / no-data error paths, and is archive-independent
/// (asserts only on the absence of the pip-refusal string, true whether or not
/// EURUSD data is on the machine).
#[test]
fn run_real_sidecar_symbol_never_hits_the_pip_refusal() {
    // A bounded window (the same EURUSD 2024-06 calendar month the gated sweep tests
    // use) keeps this fast; the full archive is unbounded and not needed for a
    // pip-lookup property. `run` is blueprint-only now (#159 cut 4).
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args([
            "run", "examples/r_sma.json", "--real", "EURUSD",
            "--from", &EURUSD_JUN2024_FROM_MS.to_string(),
            "--to", &EURUSD_JUN2024_TO_MS.to_string(),
        ])
        .output()
        .expect("spawn aura");
    let stderr = String::from_utf8_lossy(&out.stderr);
    assert!(
        !stderr.contains("no recorded geometry"),
        "a symbol with a recorded sidecar must clear the pip lookup, never the pip-refusal; stderr: {stderr}"
    );
    // It resolves to exactly one of the two legitimate outcomes: a clean report
    // (exit 0, data present) or the distinct no-local-data refusal (exit 1).
    match out.status.code() {
        Some(0) => assert!(
            String::from_utf8_lossy(&out.stdout)
                .trim_start()
                .starts_with("{\"manifest\":"),
            "exit 0 must carry a JSON report, got: {:?}",
            String::from_utf8_lossy(&out.stdout)
        ),
        Some(1) => assert!(
            stderr.contains("no local data for symbol 'EURUSD'"),
            "exit 1 for a symbol with a recorded sidecar must be the no-data path, got: {stderr}"
        ),
        other => panic!("unexpected exit for a real run with a recorded sidecar: {other:?}; stderr: {stderr}"),
    }
}

/// Property: a recorded-sidecar symbol threads its looked-up pip all the way to the
/// binary's emitted manifest — `aura run --real GER40` renders the INDEX pip
/// `sim-optimal(pip_size=1)`, not the FX `0.0001` literal the synthetic path uses.
/// This is the #22 headline at the built-binary boundary (criterion 2: GER40=1.0
/// vs EURUSD=0.0001 vs no-geometry→refuse): the metadata channel reaches stdout, so
/// equity is honestly scaled per instrument. Gated on local GER40 data — skip
/// cleanly when the archive is absent (the project's skip-on-no-data convention),
/// so it never fails on a data-less machine; the no-data path is the distinct
/// "no local data" refusal (exit 1), not the pip-refusal, asserted above.
#[test]
fn run_real_sidecar_index_pip_reaches_the_emitted_manifest() {
    // The GER40 Sept-2024 window (inclusive Unix-ms), the same calendar
    // month the gated ingest path drives; bounding it keeps the run fast.
    const FROM_MS: &str = "1725148800000";
    const TO_MS: &str = "1727740799999";
    // `run` is blueprint-only now (#159 cut 4); the shipped r-sma example still
    // routes through the same real-data pip lookup + manifest stamping.
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args(["run", "examples/r_sma.json", "--real", "GER40", "--from", FROM_MS, "--to", TO_MS])
        .output()
        .expect("spawn aura");

    // Skip on a data-less machine: a recorded-sidecar symbol with no local data takes
    // the distinct no-data path (exit 1, "no local data"), never the pip-refusal.
    if out.status.code() == Some(1) {
        let stderr = String::from_utf8_lossy(&out.stderr);
        assert!(
            stderr.contains("no local data for symbol 'GER40'"),
            "exit 1 for GER40 must be the no-data path, got: {stderr}"
        );
        eprintln!("skip: no local GER40 data");
        return;
    }

    assert_eq!(out.status.code(), Some(0), "exit: {:?}", out.status);
    let stdout = String::from_utf8(out.stdout).expect("utf-8 stdout");
    let line = stdout.trim_end();
    // the looked-up index pip (1.0 -> "1") reaches the manifest, not the FX 0.0001.
    assert!(
        line.contains("\"broker\":\"sim-optimal+risk-executor(pip_size=1)\""),
        "GER40 must render the index pip in the manifest, got: {line}"
    );
}

/// Property: the authored instrument table no longer gates real runs — a symbol
/// absent from the (removed) vetted floor but carrying a recorded geometry sidecar
/// resolves its pip from the sidecar and runs. USDJPY (never vetted; JPY pip 0.01)
/// renders the JPY pip in the manifest, proving the pip is sourced from the provider
/// geometry, not a hand-authored table entry. Gated: skips cleanly when USDJPY has
/// no local geometry/data (no sidecar → "no recorded geometry"; sidecar but no bars
/// → "no local data").
#[test]
fn run_real_nonvetted_symbol_resolves_pip_from_sidecar() {
    // FX trades continuously; the GER40 Sept-2024 window also has USDJPY bars.
    const FROM_MS: &str = "1725148800000";
    const TO_MS: &str = "1727740799999";
    // `run` is blueprint-only now (#159 cut 4); the shipped r-sma example still
    // routes through the same real-data pip lookup.
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args(["run", "examples/r_sma.json", "--real", "USDJPY", "--from", FROM_MS, "--to", TO_MS])
        .output()
        .expect("spawn aura");

    // Skip on a host without USDJPY geometry/data — either refusal is a clean skip,
    // sourced from recorded geometry, never an authored-table miss.
    if out.status.code() == Some(1) {
        let stderr = String::from_utf8_lossy(&out.stderr);
        if stderr.contains("no recorded geometry") || stderr.contains("no local data") {
            eprintln!("skip: no local USDJPY geometry/data");
            return;
        }
    }

    assert_eq!(out.status.code(), Some(0), "USDJPY should resolve and run; exit: {:?}, stderr: {}",
               out.status, String::from_utf8_lossy(&out.stderr));
    let stdout = String::from_utf8(out.stdout).expect("utf-8 stdout");
    assert!(
        stdout.contains("pip_size=0.01"),
        "USDJPY must render the JPY sidecar pip (0.01) in the manifest, got: {}",
        stdout.trim_end()
    );
}

/// `aura run <blueprint.json>` loads a serialized signal blueprint and runs it
/// end-to-end (#165): exit 0, a RunReport whose manifest carries the topology_hash.
#[test]
fn aura_run_loads_and_runs_a_blueprint_file() {
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args(["run", "examples/r_sma.json"])
        .output()
        .expect("spawn aura run blueprint");
    assert_eq!(
        out.status.code(),
        Some(0),
        "exit: {:?} stderr={}",
        out.status,
        String::from_utf8_lossy(&out.stderr)
    );
    let stdout = String::from_utf8(out.stdout).expect("utf-8");
    assert!(stdout.contains("\"topology_hash\":\""), "manifest carries the topology hash: {stdout}");
}

/// A blueprint referencing a node type outside the closed vocabulary fails clean
/// at load (the #160 closed-set guard at the data-plane face; invariant 9), and
/// (#184) the failure is phrased as house-style prose — matching the `graph
/// build` presenter's wording — never the raw `UnknownNodeType` Debug leak.
#[test]
fn aura_run_rejects_an_unknown_node_blueprint() {
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args(["run", "tests/fixtures/unknown_node.json"])
        .output()
        .expect("spawn aura run unknown");
    assert_ne!(out.status.code(), Some(0), "an unknown node type must fail the run");
    let stderr = String::from_utf8(out.stderr).expect("utf-8");
    assert!(
        stderr.contains(r#"unknown node type "Nope""#),
        "house-style prose names the type: {stderr}"
    );
    assert!(
        !stderr.contains("UnknownNodeType"),
        "does not leak the Debug variant name: {stderr}"
    );
}

/// Property (#175, refuse-don't-guess): built-in-only flags on `aura run
/// <blueprint.json>` are REFUSED, not silently dropped. The loaded-blueprint
/// grammar accepts only `--params`/`--seed`/`--real`/`--from`/`--to`; the removed
/// hand-parser rejected any other flag (exit 2). clap's optional `[blueprint]`
/// positional makes `--trace` structurally parseable (it stays on `RunCmd` for
/// the built-in-harness grammar), so the dispatch guard must re-reject it —
/// mirroring the sweep/mc blueprint branches, which reject their non-branch
/// flags exhaustively. A regression that dropped the guard would silently
/// no-op a user-provided flag. (The cost flags this test used to also cover
/// were removed from `RunCmd` entirely in #159's builtin retirement; clap now
/// rejects them as unknown args before dispatch ever sees them, so they no
/// longer exercise this guard.)
#[test]
fn run_blueprint_rejects_builtin_only_flags_exit_two() {
    let args = ["run", "examples/r_sma.json", "--trace", "foo"];
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args(args)
        .output()
        .expect("spawn aura run blueprint + builtin-only flag");
    assert_eq!(
        out.status.code(),
        Some(2),
        "{args:?} must be refused (exit 2), got {:?}; stderr: {}",
        out.status,
        String::from_utf8_lossy(&out.stderr)
    );
    assert!(
        out.stdout.is_empty(),
        "{args:?} must not emit a report on stdout: {:?}",
        String::from_utf8_lossy(&out.stdout)
    );
}

// Note (#159 cut 4 collateral): `--trace` on a single `aura run`/`aura sweep`
// is no longer reachable from any CLI surface — the built-in default that
// accepted it retired with the last two PIP strategies, and the blueprint
// branches (`dispatch_run`, `run_blueprint_sweep`) never wire a trace-persist
// path. Repointing would require new production code (out of scope for a
// test-fixing task), so the single-run trace-persist + shape tests are
// deleted here rather than adjusted.

/// Property: an unknown `chart <name>` is a runtime failure pinned at BOTH the exit code
/// and the user-facing message — exit 1 (never a panic) AND the NotFound branch's
/// "no recorded run or family '<name>'" wording, which now covers families too. A
/// regression that drifted the message (e.g. back to the run-only phrasing) or
/// downgraded the exit would slip past the bundled exit-only check; this pins the
/// message/exit contract. Archive-local: charts into an empty cwd so the name is
/// genuinely absent.
#[test]
fn chart_unknown_name_refuses_with_exit_1_and_message() {
    let cwd = temp_cwd("chart-unknown");

    let out = Command::new(BIN).args(["chart", "ghost"]).current_dir(&cwd).output().expect("spawn chart ghost");
    assert_eq!(out.status.code(), Some(1), "unknown name must exit 1: {:?}", out.status);
    assert!(out.stdout.is_empty(), "the refusal must not leak a chart page to stdout, got: {:?}", String::from_utf8_lossy(&out.stdout));
    let stderr = String::from_utf8_lossy(&out.stderr);
    assert!(
        stderr.contains("no recorded run or family 'ghost'"),
        "expected the unknown-name (run-or-family) refusal message, got: {stderr}"
    );

    let _ = std::fs::remove_dir_all(&cwd);
}

/// Property (milestone fieldtest B1): a sweep / walk-forward `--trace` family is
/// chartable by the very handle the producing run prints. Those verbs fan a
/// selection-free sweep out per member into the #224 depth-2 layout
/// `runs/traces/<name>/<cell>/<member>/index.json` (one directory level deeper
/// than the campaign path's depth-1 nominee layout `<name>/<cell>/index.json`).
/// `aura chart <handle>` must resolve that fan-out as a family: exit 0 with the
/// chart page on stdout, covering the members' taps. AND the genuinely-absent-
/// handle refusal must not tell the user to re-run a command with that same
/// handle as the `--trace` argument — that command produced the data (it would
/// already be on disk), so pointing back at it is a dead-end remedy.
///
/// The fixture is fabricated inline through `TraceStore` (no archive, no
/// data-server, no sweep run): two members under one cell, exactly the depth-2
/// shape `persist_campaign_traces`' member fan-out writes. On today's tree the
/// depth-1-only `name_kind` / `read_family` classify the handle as NotFound, so
/// `chart` exits 1 with the misleading remedy — RED on both counts.
#[test]
fn chart_opens_a_sweep_trace_family_by_its_printed_handle() {
    use aura_core::{Scalar, ScalarKind, Timestamp};
    use aura_engine::{ColumnarTrace, RunManifest};
    use aura_registry::TraceStore;

    let cwd = temp_cwd("chart-sweep-family");

    let manifest = || RunManifest {
        commit: "c0ffee".to_string(),
        params: vec![("fast".to_string(), Scalar::f64(2.0))],
        defaults: vec![],
        window: (Timestamp(1), Timestamp(2)),
        seed: 0,
        broker: "sim-optimal(pip_size=1)".to_string(),
        selection: None,
        instrument: Some("GER40".to_string()),
        topology_hash: None,
        project: None,
    };
    let taps = |a: f64, b: f64| {
        vec![ColumnarTrace::from_rows(
            "equity",
            &[ScalarKind::F64],
            &[(Timestamp(1), vec![Scalar::f64(a)]), (Timestamp(2), vec![Scalar::f64(b)])],
        )]
    };

    // The #224 sweep fan-out layout: two members under ONE cell, so each member's
    // index.json lives at depth 2 (`<name>/<cell>/<member>/index.json`) — not the
    // depth-1 nominee layout (`<name>/<cell>/index.json`) the campaign path writes.
    // `TraceStore::open` roots at `<runs_dir>/traces`, matching `env.trace_store()`.
    let store = TraceStore::open(cwd.join("runs"));
    store.write("swp/cell0/m0", &manifest(), &taps(0.0, 0.4)).expect("write member m0");
    store.write("swp/cell0/m1", &manifest(), &taps(0.0, 0.7)).expect("write member m1");

    // (1) chart the family by the handle the sweep prints: exit 0 + chart page.
    let out = Command::new(BIN).args(["chart", "swp"]).current_dir(&cwd).output().expect("spawn chart swp");
    assert_eq!(
        out.status.code(),
        Some(0),
        "charting a sweep --trace family by its handle must exit 0; got {:?}, stderr: {}",
        out.status.code(),
        String::from_utf8_lossy(&out.stderr),
    );
    let stdout = String::from_utf8_lossy(&out.stdout);
    assert!(
        stdout.contains("window.AURA_TRACES") && stdout.contains("<title>aura chart"),
        "the family handle must render a chart page covering its members' taps \
         (AURA_TRACES + chart title); stdout len {}",
        stdout.len(),
    );

    // (2) the genuinely-absent-handle refusal must not suggest re-running a
    // command with the handle itself as the `--trace` argument (a dead-end remedy).
    let ghost = Command::new(BIN).args(["chart", "ghost"]).current_dir(&cwd).output().expect("spawn chart ghost");
    assert_eq!(ghost.status.code(), Some(1), "an absent handle still exits 1: {:?}", ghost.status);
    let gerr = String::from_utf8_lossy(&ghost.stderr);
    assert!(
        !gerr.contains("--trace ghost"),
        "the not-found remedy must not tell the user to re-run with the handle as --trace \
         (that command produced the data / would not create this handle); got: {gerr}",
    );

    let _ = std::fs::remove_dir_all(&cwd);
}

/// Property (#238): a sweep / walk-forward `--trace <NAME>` family is chartable by
/// the very NAME the user chose — not only by the content-derived handle the run
/// prints. `--trace <NAME>` lands `NAME` in the generated campaign document's
/// `name` field (content-addressed via `put_campaign`), while the trace directory
/// stays keyed by the content-derived handle `{campaign8}-{run}` that
/// `append_campaign_run` re-derives (C18, untouched). Today `aura chart <NAME>`
/// classifies `NAME` as an unknown trace handle and exits 1, so the chosen name
/// has no charting effect. It must instead resolve `NAME` against the recorded
/// campaign runs: the ONE record whose stored campaign document carries
/// `name == NAME` and a persisted `trace_name` charts that family by its handle —
/// exit 0 with the same chart page charting the handle directly renders.
///
/// The state is fabricated in-process, no archive / sweep run: a campaign
/// document carrying a distinctive `name` is stored via `put_campaign`; a
/// campaign-run record with a `Some` trace_name sentinel is appended (which
/// re-derives the handle `{campaign8}-0`); a family is written under that exact
/// handle via `TraceStore` — mirroring the handle-charted sibling above (the
/// #224 depth-2 fan-out). The registry roots where `env.registry()` does
/// (`<cwd>/runs/runs.jsonl`), so the running binary reads the same store. On
/// today's tree `chart <NAME>` hits `emit_chart`'s NotFound arm (no registry
/// query), so it exits 1 with the "no recorded run or family" message — RED for
/// the right reason (the name-resolution is absent, not a fabrication error).
#[test]
fn chart_resolves_a_trace_family_by_its_chosen_campaign_name() {
    use aura_core::{Scalar, ScalarKind, Timestamp};
    use aura_engine::{ColumnarTrace, RunManifest};
    use aura_registry::{derive_trace_name, CampaignRunRecord, Registry, TraceStore};

    let cwd = temp_cwd("chart-by-campaign-name");

    // The chosen family name (`--trace <NAME>`) — distinctive, not a trace handle
    // (no such directory under runs/traces, so `name_kind` classifies NotFound).
    const CAMPAIGN_NAME: &str = "mom-ger40-sept24";

    // (1) Store a campaign document carrying that name (a valid CampaignDoc, so a
    // resolver can parse it by any strategy — Value lookup or full deserialize).
    // `put_campaign` self-keys by content, returning the id the campaign-run
    // record must reference. The registry roots where `env.registry()` does:
    // `<cwd>/runs/runs.jsonl` (campaign docs at `<cwd>/runs/campaigns/<id>.json`,
    // campaign_runs at `<cwd>/runs/campaign_runs.jsonl`).
    let reg = Registry::open(cwd.join("runs").join("runs.jsonl"));
    let campaign_json = r#"{
      "format_version": 1,
      "kind": "campaign",
      "name": "__NAME__",
      "data": { "instruments": ["GER40"],
                "windows": [ { "from_ms": 1, "to_ms": 2 } ] },
      "strategies": [
        { "ref": { "content_id": "9f3a" },
          "axes": { "fast": { "kind": "I64", "values": [8, 12] } } }
      ],
      "process": { "ref": { "content_id": "4e2d" } },
      "seed": 1,
      "presentation": { "persist_taps": ["equity"], "emit": ["family_table"] }
    }"#
    .replace("__NAME__", CAMPAIGN_NAME);
    let campaign_id = reg.put_campaign(&campaign_json).expect("store campaign document");

    // (2) Append a campaign-run record whose trace_name is Some (the executor's
    // claim sentinel; `append_campaign_run` assigns the per-campaign run counter
    // and re-derives the stored trace_name to `{campaign8}-{run}`).
    let record = CampaignRunRecord {
        campaign: campaign_id.clone(),
        process: "4e2d".to_string(),
        run: 0, // ignored — append assigns the counter
        seed: 1,
        cells: vec![],
        generalizations: vec![],
        trace_name: Some("claim".to_string()),
    };
    let run = reg.append_campaign_run(&record).expect("append campaign run");
    let handle = derive_trace_name(&campaign_id, run);

    // (3) Write a family under that exact handle via TraceStore — the same
    // fabrication the handle-charted sibling uses (#224 depth-2: two members under
    // one cell). Charting `handle` directly is thus known to render a family chart.
    let manifest = || RunManifest {
        commit: "c0ffee".to_string(),
        params: vec![("fast".to_string(), Scalar::f64(2.0))],
        defaults: vec![],
        window: (Timestamp(1), Timestamp(2)),
        seed: 0,
        broker: "sim-optimal(pip_size=1)".to_string(),
        selection: None,
        instrument: Some("GER40".to_string()),
        topology_hash: None,
        project: None,
    };
    let taps = |a: f64, b: f64| {
        vec![ColumnarTrace::from_rows(
            "equity",
            &[ScalarKind::F64],
            &[(Timestamp(1), vec![Scalar::f64(a)]), (Timestamp(2), vec![Scalar::f64(b)])],
        )]
    };
    let store = TraceStore::open(cwd.join("runs"));
    store.write(&format!("{handle}/cell0/m0"), &manifest(), &taps(0.0, 0.4)).expect("write member m0");
    store.write(&format!("{handle}/cell0/m1"), &manifest(), &taps(0.0, 0.7)).expect("write member m1");

    // Chart by the chosen NAME (not the handle): resolution must go through the
    // campaign-run query and land on `handle`, rendering the same family chart.
    let out = Command::new(BIN).args(["chart", CAMPAIGN_NAME]).current_dir(&cwd).output().expect("spawn chart <name>");
    assert_eq!(
        out.status.code(),
        Some(0),
        "charting a --trace family by its chosen campaign name must exit 0; got {:?}, stderr: {}",
        out.status.code(),
        String::from_utf8_lossy(&out.stderr),
    );
    let stdout = String::from_utf8_lossy(&out.stdout);
    assert!(
        stdout.contains("window.AURA_TRACES") && stdout.contains("<title>aura chart"),
        "the chosen name must render the same family chart as its handle \
         (AURA_TRACES + chart title); stdout len {}",
        stdout.len(),
    );

    let _ = std::fs::remove_dir_all(&cwd);
}

/// Property (#238 ambiguity): when TWO recorded campaign runs' stored documents
/// both carry the SAME chosen `name`, `chart <NAME>` must refuse (exit 1) rather
/// than silently pick one — the resolver names both candidate trace-store
/// handles in the refusal (deterministic file/append order), so the user can
/// re-run `chart` against the specific handle they mean. Mirrors the fixture of
/// [`chart_resolves_a_trace_family_by_its_chosen_campaign_name`], duplicated
/// once with a second `put_campaign` + `append_campaign_run` under the same
/// `name` (a distinct campaign document — the `data.windows` bound differs —
/// still resolves to the same `name`, exactly the "two separate --trace <NAME>
/// invocations reusing a name" scenario this guards against).
#[test]
fn chart_by_campaign_name_refuses_when_the_name_is_ambiguous() {
    use aura_core::{Scalar, ScalarKind, Timestamp};
    use aura_engine::{ColumnarTrace, RunManifest};
    use aura_registry::{derive_trace_name, CampaignRunRecord, Registry, TraceStore};

    let cwd = temp_cwd("chart-by-name-ambiguous");
    const CAMPAIGN_NAME: &str = "reused-name";

    let reg = Registry::open(cwd.join("runs").join("runs.jsonl"));
    let campaign_json_of = |to_ms: i64| {
        format!(
            r#"{{
              "format_version": 1,
              "kind": "campaign",
              "name": "{CAMPAIGN_NAME}",
              "data": {{ "instruments": ["GER40"],
                        "windows": [ {{ "from_ms": 1, "to_ms": {to_ms} }} ] }},
              "strategies": [
                {{ "ref": {{ "content_id": "9f3a" }},
                  "axes": {{ "fast": {{ "kind": "I64", "values": [8, 12] }} }} }}
              ],
              "process": {{ "ref": {{ "content_id": "4e2d" }} }},
              "seed": 1,
              "presentation": {{ "persist_taps": ["equity"], "emit": ["family_table"] }}
            }}"#
        )
    };

    let manifest = || RunManifest {
        commit: "c0ffee".to_string(),
        params: vec![("fast".to_string(), Scalar::f64(2.0))],
        defaults: vec![],
        window: (Timestamp(1), Timestamp(2)),
        seed: 0,
        broker: "sim-optimal(pip_size=1)".to_string(),
        selection: None,
        instrument: Some("GER40".to_string()),
        topology_hash: None,
        project: None,
    };
    let taps = |a: f64, b: f64| {
        vec![ColumnarTrace::from_rows(
            "equity",
            &[ScalarKind::F64],
            &[(Timestamp(1), vec![Scalar::f64(a)]), (Timestamp(2), vec![Scalar::f64(b)])],
        )]
    };
    let store = TraceStore::open(cwd.join("runs"));

    let mut handles = Vec::new();
    for to_ms in [2, 3] {
        let campaign_id = reg.put_campaign(&campaign_json_of(to_ms)).expect("store campaign document");
        let record = CampaignRunRecord {
            campaign: campaign_id.clone(),
            process: "4e2d".to_string(),
            run: 0,
            seed: 1,
            cells: vec![],
            generalizations: vec![],
            trace_name: Some("claim".to_string()),
        };
        let run = reg.append_campaign_run(&record).expect("append campaign run");
        let handle = derive_trace_name(&campaign_id, run);
        store.write(&format!("{handle}/cell0/m0"), &manifest(), &taps(0.0, 0.4)).expect("write member m0");
        store.write(&format!("{handle}/cell0/m1"), &manifest(), &taps(0.0, 0.7)).expect("write member m1");
        handles.push(handle);
    }

    let out = Command::new(BIN).args(["chart", CAMPAIGN_NAME]).current_dir(&cwd).output().expect("spawn chart <name>");
    assert_eq!(
        out.status.code(),
        Some(1),
        "an ambiguous campaign name must refuse (exit 1), not silently pick one; got {:?}, stdout: {}",
        out.status.code(),
        String::from_utf8_lossy(&out.stdout),
    );
    assert!(out.stdout.is_empty(), "the refusal must not leak a chart page to stdout");
    let stderr = String::from_utf8_lossy(&out.stderr);
    for handle in &handles {
        assert!(
            stderr.contains(handle.as_str()),
            "the refusal must list every candidate handle deterministically; got: {stderr}, want handle: {handle}"
        );
    }

    let _ = std::fs::remove_dir_all(&cwd);
}

#[test]
fn no_args_prints_usage_and_exits_two() {
    let out = Command::new(BIN).output().expect("spawn aura");
    assert_eq!(out.status.code(), Some(2), "exit status: {:?}", out.status);
    assert!(out.stdout.is_empty(), "stdout should be empty on the usage path");
    let stderr = String::from_utf8(out.stderr).expect("utf-8 stderr");
    assert!(stderr.contains("Usage"), "stderr was: {stderr:?}");
}

/// Property: `--help`/`-h` is the success-path help affordance, not the error
/// path. A newcomer's reflex first command (C22 first-contact ergonomics, #20)
/// prints the usage to stdout and exits 0; the conventional `-h` alias behaves
/// identically. An *unknown* subcommand keeps the error path (exit 2) so the
/// help fix does not regress bad-args handling.
#[test]
fn help_flag_prints_usage_to_stdout_and_exits_zero() {
    for flag in ["--help", "-h"] {
        let out = Command::new(BIN).arg(flag).output().expect("spawn aura --help");
        assert_eq!(out.status.code(), Some(0), "`aura {flag}` exit: {:?}", out.status);
        let stdout = String::from_utf8(out.stdout).expect("utf-8 stdout");
        assert!(
            !stdout.trim().is_empty(),
            "`aura {flag}` should print help to stdout, got: {stdout:?}"
        );
        // The usage names the only subcommand a newcomer can run.
        assert!(
            stdout.contains("run"),
            "`aura {flag}` help should mention the `run` subcommand, got: {stdout:?}"
        );
    }

    // Negative-preservation: an unknown subcommand is still the error path.
    let out = Command::new(BIN).arg("frobnicate").output().expect("spawn aura frobnicate");
    assert_eq!(
        out.status.code(),
        Some(2),
        "unknown subcommand must stay exit 2: {:?}",
        out.status
    );
}

/// Property (post-clap migration, #175): subcommand help is SCOPED — `aura <sub>
/// --help` (and `-h`) prints that subcommand's own options to stdout and exits 0,
/// with nothing on stderr, for every subcommand. This supersedes the retired #131
/// "uniform (byte-identical) help" surface: clap gives each subcommand its own
/// scoped Options section rather than one shared global usage blob.
#[test]
fn per_subcommand_help_is_scoped_stdout_exit_zero() {
    for sub in ["run", "chart", "sweep", "walkforward", "mc", "graph", "runs"] {
        for help in ["--help", "-h"] {
            let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
                .args([sub, help]).output().expect("spawn");
            assert_eq!(out.status.code(), Some(0), "{sub} {help} exit");
            assert!(!out.stdout.is_empty(), "{sub} {help} stdout empty");
            assert!(out.stderr.is_empty(), "{sub} {help} stderr not empty: {:?}",
                String::from_utf8_lossy(&out.stderr));
        }
    }
}

/// Property: `aura graph` emits a single self-contained HTML page — the
/// interactive pin-graph viewer with the deterministic model inlined and the
/// vendored Graphviz-WASM + pan-zoom inlined (no remote fetch). Driven through
/// the built binary so it pins the observable CLI contract. The DOT/SVG are
/// Graphviz-version-dependent and not asserted (the prototype is the by-hand
/// visual reference); this pins the page envelope + the retirement of the old
/// ascii-dag notation.
#[test]
fn graph_emits_self_contained_html_viewer() {
    let out = Command::new(BIN).arg("graph").output().expect("spawn aura graph");
    assert_eq!(out.status.code(), Some(0), "`aura graph` exit: {:?}", out.status);
    let stdout = String::from_utf8(out.stdout).expect("utf-8 stdout");
    // a self-contained HTML document...
    assert!(
        stdout.trim_start().starts_with("<!doctype html>"),
        "not an HTML doc: {:?}",
        &stdout[..stdout.len().min(80)]
    );
    // ...with the deterministic model inlined as the viewer's data source...
    assert!(stdout.contains("window.AURA_MODEL = {\"root\":"), "model not inlined: {stdout:?}");
    // ...the Graphviz-WASM bootstrap present, and no remote asset fetch.
    assert!(stdout.contains("Viz.instance"), "viewer bootstrap missing");
    assert!(!stdout.contains("<script src"), "assets must be inlined, found remote <script src>");
    // the retired ascii-dag invented notation is gone.
    assert!(!stdout.contains("Sub(#Sf,#Ss)"), "retired ascii-dag notation must be gone: {stdout}");
}

/// Property (#28): `aura graph <blueprint.json>` renders the CONSUMER's own
/// graph — the structure of the file it is handed — not the embedded r_sma
/// sample. Mirrors the `dispatch_run` dual-grammar: a first-positional naming
/// an existing `.json` selects the loaded-blueprint branch, which reads the
/// file, reloads it via `blueprint_from_json`, and `render_html`s it — so the
/// emitted page's inlined `window.AURA_MODEL` reflects THAT file's nodes.
/// Handed the retired r-breakout example (std-vocabulary only, so no project is
/// needed), the page must carry a marker unique to that graph — `channel_hi`, a
/// RollingMax node name absent from the sample and from every static viewer
/// asset — and must NOT carry the sample-only `"type":"Bias"` node, i.e. it did
/// not silently fall back to the embedded r_sma sample. Driven through the built
/// binary so it pins the observable CLI contract.
#[test]
fn graph_renders_a_consumer_blueprint_file_not_the_embedded_sample() {
    let cwd = temp_cwd("graph-consumer-blueprint");
    // A committed consumer blueprint that is NOT the embedded sample and resolves
    // against the base std vocabulary alone (RollingMax/RollingMin/Delay/Gt/Latch/
    // Sub) — no project fixture required. Absolute path, so the temp cwd is free of
    // any Aura.toml (guarantees std-env resolution).
    let blueprint = format!("{}/examples/r_breakout.json", env!("CARGO_MANIFEST_DIR"));
    let out = Command::new(BIN)
        .arg("graph")
        .arg(&blueprint)
        .current_dir(&cwd)
        .output()
        .expect("spawn aura graph <blueprint>");
    assert_eq!(
        out.status.code(),
        Some(0),
        "`aura graph <blueprint.json>` exit: {:?}\nstderr: {}",
        out.status,
        String::from_utf8_lossy(&out.stderr)
    );
    let stdout = String::from_utf8(out.stdout).expect("utf-8 stdout");
    // the page inlines a model as the viewer's data source...
    assert!(
        stdout.contains("window.AURA_MODEL = {\"root\":"),
        "model not inlined into the page"
    );
    // ...and that model is the CONSUMER graph's — its own node `channel_hi`
    // (unique to r-breakout, absent from the r_sma sample and from every static
    // asset) is present.
    assert!(
        stdout.contains("channel_hi"),
        "the consumer blueprint's own node (channel_hi) is missing — its graph was not rendered"
    );
    // ...NOT the embedded r_sma sample: the sample's Bias node type must be absent
    // (proof the file was rendered, not silently ignored in favour of the sample).
    assert!(
        !stdout.contains("\"type\":\"Bias\""),
        "rendered the embedded r_sma sample (its Bias node is present) instead of the consumer blueprint"
    );
}

/// Property (#28 close-audit): a NAMED-but-unreadable blueprint arg to
/// `aura graph` is a usage fault, not a silent fallback to the embedded sample.
/// `dispatch_graph`'s no-subcommand arm branches on `is_blueprint_file`, which
/// returns `None` for BOTH "no positional given" (→ legitimate sample default)
/// and "a positional was given but it is not a readable `.json` file" (→ a bad
/// arg). The two must not be conflated: a bad arg (typo, nonexistent path, wrong
/// extension) must `eprintln!` a usage error and exit 2, mirroring `dispatch_run`
/// — NOT render the r_sma sample and exit 0, which re-introduces the exact #28
/// friction ("I asked for my graph, got the sample") in the error path. The
/// bare-`aura graph` default (no positional → sample, exit 0) is guarded in the
/// same test so the fix cannot regress it.
#[test]
fn graph_with_bad_blueprint_arg_is_a_usage_fault_not_a_silent_sample() {
    let cwd = temp_cwd("graph-bad-arg");
    // A named .json path that does not exist: `is_blueprint_file` returns None,
    // and the buggy arm falls through to the embedded sample instead of refusing.
    let out = Command::new(BIN)
        .arg("graph")
        .arg("does-not-exist.json")
        .current_dir(&cwd)
        .output()
        .expect("spawn aura graph <bad-arg>");
    // Headline pin: a named-but-unreadable blueprint arg exits 2 (usage fault),
    // mirroring `dispatch_run`'s bad-blueprint branch — not 0.
    assert_eq!(
        out.status.code(),
        Some(2),
        "`aura graph <nonexistent.json>` must be a usage fault (exit 2), not a silent sample fallback; got {:?}\nstderr: {}",
        out.status,
        String::from_utf8_lossy(&out.stderr),
    );
    // ...and it must NOT emit the embedded r_sma sample: the sample's inlined
    // model + its Bias node must be absent from stdout.
    let stdout = String::from_utf8(out.stdout).expect("utf-8 stdout");
    assert!(
        !stdout.contains("\"type\":\"Bias\""),
        "a bad blueprint arg silently rendered the embedded r_sma sample (its Bias node is present) instead of refusing"
    );

    // Guard the legitimate default: bare `aura graph` (no positional) still
    // renders the embedded sample and exits 0.
    let default = Command::new(BIN)
        .arg("graph")
        .current_dir(&cwd)
        .output()
        .expect("spawn aura graph");
    assert_eq!(
        default.status.code(),
        Some(0),
        "bare `aura graph` must still render the sample (exit 0); got {:?}",
        default.status,
    );
    let default_out = String::from_utf8(default.stdout).expect("utf-8 stdout");
    assert!(
        default_out.contains("\"type\":\"Bias\""),
        "bare `aura graph` must render the embedded r_sma sample (its Bias node), the no-positional default"
    );
}

#[test]
fn sweep_with_trailing_token_is_strict_and_exits_two() {
    // strict-arg reading (#16): an unexpected trailing token is a usage error.
    let out = Command::new(BIN).args(["sweep", "extra"]).output().expect("spawn aura");
    assert_eq!(out.status.code(), Some(2), "exit status: {:?}", out.status);
}

#[test]
fn runs_families_list_and_per_family_rank_across_invocations() {
    let cwd = temp_cwd("runs-flow");
    // two blueprint sweeps accumulate two families (sweep-0, sweep-1) over time,
    // each a related set of 4 records (C18/C21 lineage), in the sibling family
    // store. `sweep` is blueprint-only now (#159 cut 4); the shipped open r-sma
    // example over a 2x2 axis grid stands in for the retired built-in default.
    let fixture = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    for _ in 0..2 {
        let out = Command::new(BIN)
            .args([
                "sweep", &fixture,
                "--axis", "sma_signal.fast.length=2,4",
                "--axis", "sma_signal.slow.length=8,16",
            ])
            .current_dir(&cwd)
            .output()
            .expect("spawn sweep");
        assert!(out.status.success(), "sweep exit: {:?}; stderr: {}", out.status, String::from_utf8_lossy(&out.stderr));
    }

    // `runs families` shows both stored families in first-seen order, each with
    // its kind + member count.
    let fams = Command::new(BIN).args(["runs", "families"]).current_dir(&cwd).output().expect("spawn families");
    assert!(fams.status.success(), "families exit: {:?}", fams.status);
    let fams_out = String::from_utf8(fams.stdout).expect("utf-8");
    let fam_lines: Vec<&str> = fams_out.lines().collect();
    assert_eq!(fam_lines.len(), 2, "families: {fams_out:?}");
    assert!(fam_lines[0].contains("\"family_id\":\"sweep-0\""), "fam0: {fams_out:?}");
    assert!(fam_lines[1].contains("\"family_id\":\"sweep-1\""), "fam1: {fams_out:?}");
    assert!(fam_lines.iter().all(|l| l.contains("\"members\":4")), "members: {fams_out:?}");

    // `runs family sweep-0 rank total_pips`: the family's 4 members, highest
    // total_pips first.
    let rank = Command::new(BIN)
        .args(["runs", "family", "sweep-0", "rank", "total_pips"])
        .current_dir(&cwd)
        .output()
        .expect("spawn rank");
    assert!(rank.status.success(), "rank exit: {:?}", rank.status);
    let rank_out = String::from_utf8(rank.stdout).expect("utf-8");
    assert_eq!(rank_out.lines().count(), 4, "rank: {rank_out:?}");
    assert!(rank_out.lines().all(|l| l.starts_with("{\"manifest\":{")), "rank shape: {rank_out:?}");
    let pips: Vec<f64> = rank_out
        .lines()
        .map(|l| {
            let key = "\"total_pips\":";
            let start = l.find(key).expect("line has total_pips") + key.len();
            let tail = &l[start..];
            let end = tail.find([',', '}']).expect("token end");
            tail[..end].parse().expect("total_pips is an f64")
        })
        .collect();
    assert!(pips.windows(2).all(|w| w[0] >= w[1]), "rank not descending: {pips:?}");

    // an unknown metric is a usage error
    let bogus = Command::new(BIN)
        .args(["runs", "family", "sweep-0", "rank", "bogus"])
        .current_dir(&cwd)
        .output()
        .expect("spawn bogus");
    assert_eq!(bogus.status.code(), Some(2), "bogus exit: {:?}", bogus.status);

    // an unknown family id is an empty family: zero lines, exit 0.
    let unknown = Command::new(BIN)
        .args(["runs", "family", "nope-9"])
        .current_dir(&cwd)
        .output()
        .expect("spawn unknown");
    assert!(unknown.status.success(), "unknown family exit: {:?}", unknown.status);
    assert_eq!(unknown.stdout.len(), 0, "unknown family stdout: {:?}", unknown.stdout);

    let _ = std::fs::remove_dir_all(&cwd);
}

#[test]
fn runs_list_and_rank_are_retired_and_exit_two() {
    // `aura runs list` / `runs rank <metric>` were retired (#73): families
    // (sweep/mc/walkforward, C21) subsume standalone over-time comparison. The
    // dropped subcommands now fall through to the strict-arg usage error (exit 2),
    // before any registry access.
    let list = Command::new(BIN).args(["runs", "list"]).output().expect("spawn list");
    assert_eq!(list.status.code(), Some(2), "retired `runs list` must exit 2: {:?}", list.status);
    let rank = Command::new(BIN).args(["runs", "rank", "total_pips"]).output().expect("spawn rank");
    assert_eq!(rank.status.code(), Some(2), "retired `runs rank` must exit 2: {:?}", rank.status);
}

#[test]
fn runs_bare_subcommand_is_strict_and_exits_two() {
    // `aura runs` with no/unknown sub-subcommand is a usage error (#16 strict).
    let out = Command::new(BIN).arg("runs").output().expect("spawn aura runs");
    assert_eq!(out.status.code(), Some(2), "exit status: {:?}", out.status);
}

// Note: the family-trace-persistence tests deleted above (mc/sweep/walkforward
// per-member trace dirs, the momentum bool-param dir/gate proofs, the sweep
// determinism pin) are the Step 3 PIP-surface deletions — same underlying
// cause as the collateral note above (no blueprint-mode arm persists
// per-member traces).

// --- Real-data family path (#106, plan 0060 Task 5) -------------------------
//
// These drive `aura sweep|walkforward --real <SYMBOL>` end to end over the local
// data-server archive. The gated tests need the Pepperstone M1 archive present;
// they skip (print + pass) on a data-less machine, the project's
// skip-on-no-data convention. The pip-refusal test is NOT gated: a symbol with no
// recorded geometry is rejected before any data access.

// EURUSD 2024-06 (UTC inclusive ms): a bounded recorded-sidecar real window.
const EURUSD_JUN2024_FROM_MS: i64 = 1_717_200_000_000;
const EURUSD_JUN2024_TO_MS: i64 = 1_719_791_999_999;

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

/// Property: the per-instrument pip refusal fires for `aura sweep --real
/// <no-geometry symbol>` exactly as it does for `aura run --real` — a symbol with no
/// recorded geometry (`NONEXISTENT`, no sidecar on any host) is rejected with exit 1
/// and the "no recorded geometry" message BEFORE any bar-data access. NOT gated: the
/// refusal precedes the archive entirely, so it is CI-safe and runs on every machine.
/// (#159 cut 4: `sweep` is blueprint-only now, so the fixture supplies a valid
/// blueprint + axis to clear the usage grammar before the geometry refusal fires.)
#[test]
fn sweep_real_no_geometry_symbol_refuses_with_exit_1() {
    // not gated: the pip refusal happens before any data access.
    let (dir, _g) = fresh_project();
    let fixture = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let out = std::process::Command::new(BIN)
        .args(["sweep", &fixture, "--axis", "sma_signal.fast.length=2,4", "--real", "NONEXISTENT"])
        .current_dir(&dir).output().unwrap();
    assert_eq!(out.status.code(), Some(1));
    assert!(String::from_utf8_lossy(&out.stderr).contains("no recorded geometry"));
}

// GER40 Sept-2024 window (inclusive Unix-ms) — the same gated calendar month
// used by `run_real_sidecar_index_pip_reaches_the_emitted_manifest` and the
// campaign-path e2e in `research_docs.rs`.
const GER40_SEPT2024_FROM_MS: &str = "1725148800000";
const GER40_SEPT2024_TO_MS: &str = "1727740799999";

/// Property (#242): a `--real` window that holds no bars for a symbol that IS in
/// the archive refuses with a WINDOW-aware message — distinct from the genuinely
/// unknown-symbol refusal. GER40 is present (the covered Sept-2024 window runs),
/// so an *inverted* window (`--from` > `--to`) is empty by construction for any
/// archive — no bar can satisfy `from <= t <= to` — yet the symbol, its geometry,
/// and the month's file all exist. The refusal must name the requested window
/// rather than reuse the symbol-absence message "no local data for symbol", which
/// misattributes an empty WINDOW to a missing SYMBOL. Gated on the local archive;
/// skips cleanly when absent (the project's skip-on-no-data convention), so CI
/// without the archive stays green.
#[test]
fn run_real_empty_window_refusal_names_the_window_not_the_symbol() {
    if !local_data_present() {
        eprintln!("skip: no local data at {}", data_server::DEFAULT_DATA_PATH);
        return;
    }
    // The covered Sept-2024 window, INVERTED (from > to): `has_symbol("GER40")` is
    // window-independent and passes, and the month's file overlaps at date
    // granularity, so the empty result is a WINDOW fact, not symbol absence.
    let out = std::process::Command::new(BIN)
        .args([
            "run", "examples/r_sma.json",
            "--real", "GER40",
            "--from", GER40_SEPT2024_TO_MS, // inverted: from > to,
            "--to", GER40_SEPT2024_FROM_MS, // so the in-window bar set is empty
        ])
        .output()
        .expect("spawn aura");

    let stderr = String::from_utf8_lossy(&out.stderr);
    // An empty --real window is still a refusal (exit 1 unchanged); only the
    // message changes — the genuinely-unknown-symbol path keeps exit 1 too.
    assert_eq!(
        out.status.code(),
        Some(1),
        "an empty --real window still refuses with exit 1; stderr: {stderr}"
    );
    // The distinction (#242): the empty-window refusal NAMES the window and does
    // NOT reuse the wholesale symbol-absence message for a symbol that is present.
    assert!(
        stderr.to_lowercase().contains("window"),
        "empty-window refusal must name the requested window (GER40 IS present), got: {stderr}"
    );
    assert!(
        !stderr.contains("no local data for symbol"),
        "empty-window refusal must not reuse the symbol-absence message, got: {stderr}"
    );
}

/// Characterization pin (grounding-check ratification for the risk-regime-axis
/// spec): the dissolved real-data sweep member manifest stamps its resolved stop
/// (`stop_length`/`stop_k`) — the default regime here. The R-path stop knobs are
/// bound via `run_blueprint_member`'s `stop` param and stamped into the manifest
/// alongside the swept signal axes; the sibling
/// `sweep_real_blueprint_member_lines_pin_the_dissolved_contract` locates each
/// binding with `find` and so never pins the exact resolved VALUES this test
/// checks. Gated on the local GER40 archive; skips cleanly when absent.
#[test]
fn sweep_real_blueprint_member_manifest_stamps_the_default_stop() {
    if !local_data_present() {
        eprintln!("skip: no local data at {}", data_server::DEFAULT_DATA_PATH);
        return;
    }
    let (dir, _g) = fresh_project();
    let fixture = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let out = std::process::Command::new(BIN)
        .args([
            "sweep", &fixture,
            "--real", "GER40",
            "--from", GER40_SEPT2024_FROM_MS,
            "--to", GER40_SEPT2024_TO_MS,
            "--axis", "sma_signal.fast.length=2,4",
            "--axis", "sma_signal.slow.length=8",
            "--name", "brp_nostop",
        ])
        .current_dir(&dir)
        .output()
        .unwrap();
    assert!(out.status.success(), "stderr: {}", String::from_utf8_lossy(&out.stderr));
    let stdout = String::from_utf8_lossy(&out.stdout).into_owned();
    let lines: Vec<&str> = stdout.lines().collect();
    assert_eq!(lines.len(), 2, "one member line per grid point: {stdout}");

    for line in &lines {
        let v: serde_json::Value = serde_json::from_str(line).expect("member line parses as JSON");
        let params = v["report"]["manifest"]["params"]
            .as_array()
            .expect("manifest.params is an array");
        let get = |name: &str| params.iter().find(|p| p[0].as_str() == Some(name));
        assert_eq!(get("stop_length").and_then(|p| p[1]["I64"].as_i64()), Some(3),
            "the resolved default stop length is stamped: {line}");
        assert_eq!(get("stop_k").and_then(|p| p[1]["F64"].as_f64()), Some(2.0),
            "the resolved default stop k is stamped: {line}");
    }
}

/// Property (#218 migration follow-up): running a dissolved `sweep` inside a
/// project (`fresh_project()`, the fixture this migration introduced) stamps
/// `report.manifest.project` on every member with the project's audit-trail
/// provenance (C18/C16) — the `demo` namespace and a well-formed 64-hex-char
/// dylib sha256 — rather than the `None` a bare-cwd run (no `Aura.toml`)
/// produces. Before this migration no cli_run.rs sweep test ran inside a real
/// project, so this stamping path was never exercised end-to-end. Gated on the
/// local GER40 archive; skips cleanly when absent.
#[test]
fn sweep_real_blueprint_member_manifest_stamps_project_provenance() {
    if !local_data_present() {
        eprintln!("skip: no local data at {}", data_server::DEFAULT_DATA_PATH);
        return;
    }
    let (dir, _g) = fresh_project();
    let fixture = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let out = std::process::Command::new(BIN)
        .args([
            "sweep", &fixture,
            "--real", "GER40",
            "--from", GER40_SEPT2024_FROM_MS,
            "--to", GER40_SEPT2024_TO_MS,
            "--axis", "sma_signal.fast.length=2,4",
            "--axis", "sma_signal.slow.length=8",
            "--name", "brp_provenance",
        ])
        .current_dir(&dir)
        .output()
        .unwrap();
    assert!(out.status.success(), "stderr: {}", String::from_utf8_lossy(&out.stderr));
    let stdout = String::from_utf8_lossy(&out.stdout).into_owned();
    let lines: Vec<&str> = stdout.lines().collect();
    assert_eq!(lines.len(), 2, "one member line per grid point: {stdout}");

    for line in &lines {
        let v: serde_json::Value = serde_json::from_str(line).expect("member line parses as JSON");
        let project = &v["report"]["manifest"]["project"];
        assert_eq!(
            project["namespace"].as_str(), Some("demo"),
            "member manifest carries the fixture project's namespace: {line}"
        );
        let sha = project["dylib_sha256"].as_str().unwrap_or_else(|| panic!("dylib_sha256 present: {line}"));
        assert_eq!(sha.len(), 64, "dylib_sha256 is a full sha256 hex digest: {line}");
        assert!(sha.chars().all(|c| c.is_ascii_hexdigit()), "dylib_sha256 is hex: {line}");
    }
}

/// Acceptance (harness-input-binding spec, #231): an OHLC-consuming strategy
/// blueprint runs end to end on real archive data through `aura run` — the
/// role names bind columns, three real sources open in canonical order, the
/// close column feeds broker/executor, and the report carries the R summary.
/// Gated on the local GER40 archive; skips cleanly when absent.
#[test]
fn run_real_ohlc_channel_example_end_to_end() {
    if !local_data_present() {
        eprintln!("skip: no local data at {}", data_server::DEFAULT_DATA_PATH);
        return;
    }
    let cwd = temp_cwd("ohlc-run-real");
    let fixture = format!("{}/examples/r_channel.json", env!("CARGO_MANIFEST_DIR"));
    let out = std::process::Command::new(BIN)
        .args([
            "run", &fixture,
            "--real", "GER40",
            "--from", GER40_SEPT2024_FROM_MS,
            "--to", GER40_SEPT2024_TO_MS,
        ])
        .current_dir(&cwd)
        .output()
        .unwrap();
    assert!(out.status.success(), "stderr: {}", String::from_utf8_lossy(&out.stderr));
    let stdout = String::from_utf8_lossy(&out.stdout);
    let v: serde_json::Value = serde_json::from_str(stdout.trim()).expect("run report parses as JSON");
    assert!(
        v["manifest"]["topology_hash"].as_str().is_some(),
        "report carries the signal hash: {stdout}"
    );
    assert!(v["metrics"]["r"].is_object(), "the R summary is computed: {stdout}");
    let _ = std::fs::remove_dir_all(&cwd);
}

/// Property (#275, hostless): `run_signal_r`'s migrated production call site
/// (`h.run_bound(key_supply(&binding, sources))`, replacing the old positional
/// `h.run(sources)`) pairs each opened archive column with its declared role
/// **by name**, not by list position, for a genuinely multi-role strategy
/// (r_channel declares three distinct roles — high, low, close). Unlike
/// `run_real_ohlc_channel_example_end_to_end` above (gated on a local GER40
/// mount, which this sandbox does not have), this drives the same fixture over
/// `fresh_project_with_data`'s hermetic synthetic archive, so the by-name-binding
/// regression guard for this call site actually runs on every host, not just a
/// data-ful one. The exact R summary is a characterization pin against the
/// synthetic archive's deterministic price curve: a column swap (e.g. high/low
/// crossed) would change the channel breakout logic and move this number.
#[test]
fn run_real_ohlc_channel_example_end_to_end_hostless() {
    let (cwd, _g) = fresh_project_with_data();
    let fixture = format!("{}/examples/r_channel.json", env!("CARGO_MANIFEST_DIR"));
    let out = std::process::Command::new(BIN)
        .args([
            "run", &fixture,
            "--real", "SYMA",
            "--from", "1707120000000", // 2024-02-05 08:00 UTC (Monday)
            "--to", "1707325200000",   // 2024-02-07 17:00 UTC (Wednesday)
        ])
        .current_dir(&cwd)
        .output()
        .unwrap();
    assert!(out.status.success(), "stderr: {}", String::from_utf8_lossy(&out.stderr));
    let stdout = String::from_utf8_lossy(&out.stdout);
    let v: serde_json::Value = serde_json::from_str(stdout.trim()).expect("run report parses as JSON");
    assert!(
        v["manifest"]["topology_hash"].as_str().is_some(),
        "report carries the signal hash: {stdout}"
    );
    assert_eq!(
        v["metrics"]["r"]["n_trades"], serde_json::json!(1),
        "characterization pin over the deterministic synthetic archive \
         (a high/low/close column mispairing would move this count): {stdout}"
    );
    assert_eq!(
        v["metrics"]["r"]["expectancy_r"], serde_json::json!(-0.7466116470582123),
        "characterization pin over the deterministic synthetic archive \
         (a high/low/close column mispairing would move this value): {stdout}"
    );
}

/// The synthetic walk generates a close series only: a multi-column blueprint
/// without `--real` refuses honestly (exit 1, columns + remedy named). NOT
/// gated — the refusal precedes any data access.
#[test]
fn run_synthetic_refuses_a_multi_column_blueprint() {
    let fixture = format!("{}/examples/r_channel.json", env!("CARGO_MANIFEST_DIR"));
    let out = std::process::Command::new(BIN)
        .args(["run", &fixture])
        .output()
        .unwrap();
    assert_eq!(out.status.code(), Some(1));
    let stderr = String::from_utf8_lossy(&out.stderr);
    assert!(stderr.contains("consumes columns beyond close"), "names the shape: {stderr}");
    assert!(stderr.contains("--real"), "names the remedy: {stderr}");
}

/// The dissolved real-data sweep + reproduce over the OHLC example (#231
/// acceptance): `--list-axes` pins the ganged channel knob's wrapped axis
/// name, the sweep runs one member per grid point over three real columns,
/// and every member re-derives bit-identically from the store (#229 real-data
/// reproduce, now multi-column). Gated on the local GER40 archive.
#[test]
fn sweep_real_ohlc_channel_members_run_and_reproduce() {
    if !local_data_present() {
        eprintln!("skip: no local data at {}", data_server::DEFAULT_DATA_PATH);
        return;
    }
    let (dir, _g) = fresh_project();
    let fixture = format!("{}/tests/fixtures/r_channel_open.json", env!("CARGO_MANIFEST_DIR"));
    let axes = std::process::Command::new(BIN)
        .args(["sweep", &fixture, "--list-axes"])
        .current_dir(&dir)
        .output()
        .unwrap();
    assert_eq!(
        String::from_utf8_lossy(&axes.stdout),
        "hl_channel.channel_length:I64\n\
         hl_channel.prev_high.lag:I64 default=1\n\
         hl_channel.prev_low.lag:I64 default=1\n",
        "the ganged channel knob is the one open axis, alongside the fixture's \
         bound Delay lags as #246 defaults; stderr: {}",
        String::from_utf8_lossy(&axes.stderr)
    );
    let out = std::process::Command::new(BIN)
        .args([
            "sweep", &fixture,
            "--real", "GER40",
            "--from", GER40_SEPT2024_FROM_MS,
            "--to", GER40_SEPT2024_TO_MS,
            "--axis", "hl_channel.channel_length=3,5",
            "--name", "chan",
        ])
        .current_dir(&dir)
        .output()
        .unwrap();
    assert!(out.status.success(), "stderr: {}", String::from_utf8_lossy(&out.stderr));
    let stdout = String::from_utf8_lossy(&out.stdout).into_owned();
    assert_eq!(stdout.lines().count(), 2, "one member line per channel length: {stdout}");
    for line in stdout.lines() {
        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"),
            "campaign member manifest stamps the instrument: {line}"
        );
    }
    let family_id = serde_json::from_str::<serde_json::Value>(stdout.lines().next().unwrap())
        .expect("first member line parses")["family_id"]
        .as_str()
        .expect("member line carries family_id")
        .to_string();
    let rep = std::process::Command::new(BIN)
        .args(["reproduce", &family_id])
        .current_dir(&dir)
        .output()
        .unwrap();
    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"),
        "stdout: {}",
        String::from_utf8_lossy(&rep.stdout)
    );
}

/// Property (#275, hostless): `run_blueprint_member`'s migrated production call
/// site (`h.run_bound(key_supply(binding, sources))`) resolves the same
/// role-keyed pairing on every member of a multi-column sweep AND on the
/// independent `reproduce` re-run — a sibling of
/// `sweep_real_ohlc_channel_members_run_and_reproduce` above (gated on a local
/// GER40 mount this sandbox lacks) that drives the identical multi-role fixture
/// over `fresh_project_with_data`'s hermetic synthetic archive, so this call
/// site's by-name-binding regression guard runs on every host. Bit-identical
/// reproduction is the stronger check here: `reproduce` re-derives `binding` and
/// re-opens `sources` independently of the first run, so it can only match if
/// both derivations resolve to the exact same role→column pairing.
#[test]
fn sweep_real_ohlc_channel_members_run_and_reproduce_hostless() {
    let (dir, _g) = fresh_project_with_data();
    let fixture = format!("{}/tests/fixtures/r_channel_open.json", env!("CARGO_MANIFEST_DIR"));
    let out = std::process::Command::new(BIN)
        .args([
            "sweep", &fixture,
            "--real", "SYMA",
            "--from", "1707120000000", // 2024-02-05 08:00 UTC (Monday)
            "--to", "1707325200000",   // 2024-02-07 17:00 UTC (Wednesday)
            "--axis", "hl_channel.channel_length=3,5",
            "--name", "chan-hostless",
        ])
        .current_dir(&dir)
        .output()
        .unwrap();
    assert!(out.status.success(), "stderr: {}", String::from_utf8_lossy(&out.stderr));
    let stdout = String::from_utf8_lossy(&out.stdout).into_owned();
    assert_eq!(stdout.lines().count(), 2, "one member line per channel length: {stdout}");
    for line in stdout.lines() {
        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("SYMA"),
            "campaign member manifest stamps the instrument: {line}"
        );
    }
    let family_id = serde_json::from_str::<serde_json::Value>(stdout.lines().next().unwrap())
        .expect("first member line parses")["family_id"]
        .as_str()
        .expect("member line carries family_id")
        .to_string();
    let rep = std::process::Command::new(BIN)
        .args(["reproduce", &family_id])
        .current_dir(&dir)
        .output()
        .unwrap();
    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"),
        "stdout: {}",
        String::from_utf8_lossy(&rep.stdout)
    );
}

/// Property: the real-data blueprint sweep path (`aura sweep <blueprint.json>
/// --real SYM --axis …`, dispatched as sugar over the one campaign executor)
/// prints one member line per grid point, IN AXIS ORDER (odometer: the first
/// `--axis` varies slowest), each line JSON with a `family_id` and a `report`
/// whose `manifest.params` carries every swept param binding — and, the
/// sanctioned additive delta over the synthetic path, **`report.manifest`
/// carries a stamped `instrument` key** (the campaign substrate stamps every
/// member's manifest with the instrument under test, exactly like the sibling
/// `campaign_run_real_e2e_sweep_gate_walkforward` in `research_docs.rs`). Also
/// pins that the run persists exactly one new `FamilyKind::Sweep` family with
/// the grid's member count, and that the underlying generated process/campaign
/// documents and campaign-run record are durably auto-registered. Gated on the
/// local GER40 archive (the project's skip-on-no-data convention); skips
/// cleanly when absent.
#[test]
fn sweep_real_blueprint_member_lines_pin_the_dissolved_contract() {
    if !local_data_present() {
        eprintln!("skip: no local data at {}", data_server::DEFAULT_DATA_PATH);
        return;
    }
    let (dir, _g) = fresh_project();
    let fixture = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let out = std::process::Command::new(BIN)
        .args([
            "sweep", &fixture,
            "--real", "GER40",
            "--from", GER40_SEPT2024_FROM_MS,
            "--to", GER40_SEPT2024_TO_MS,
            "--axis", "sma_signal.fast.length=2,4",
            "--axis", "sma_signal.slow.length=8,16",
            "--name", "brp",
        ])
        .current_dir(&dir)
        .output()
        .unwrap();
    assert!(out.status.success(), "stderr: {}", String::from_utf8_lossy(&out.stderr));
    let stdout = String::from_utf8_lossy(&out.stdout).into_owned();
    let lines: Vec<&str> = stdout.lines().collect();
    assert_eq!(lines.len(), 4, "one member line per 2x2 grid point: {stdout}");

    // Odometer order: `sma_signal.fast.length` (first `--axis`, outer/slowest)
    // x `sma_signal.slow.length` (second `--axis`, inner) -> (2,8),(2,16),(4,8),(4,16).
    let expected: [[(&str, i64); 2]; 4] = [
        [("sma_signal.fast.length", 2), ("sma_signal.slow.length", 8)],
        [("sma_signal.fast.length", 2), ("sma_signal.slow.length", 16)],
        [("sma_signal.fast.length", 4), ("sma_signal.slow.length", 8)],
        [("sma_signal.fast.length", 4), ("sma_signal.slow.length", 16)],
    ];

    for i in 0..4 {
        let line = lines[i];
        let v: serde_json::Value = serde_json::from_str(line).expect("member line parses as JSON");
        assert!(v["family_id"].as_str().is_some(), "member line carries a family_id: {line}");
        let manifest = &v["report"]["manifest"];
        assert!(manifest.is_object(), "report.manifest is present: {line}");
        let params = manifest["params"].as_array().expect("manifest.params is an array");
        for (name, value) in expected[i] {
            let bound = params
                .iter()
                .find(|p| p[0].as_str() == Some(name))
                .and_then(|p| p[1]["I64"].as_i64());
            assert_eq!(bound, Some(value), "manifest carries the swept binding {name}={value}: {line}");
        }
        // the sanctioned additive delta: the campaign substrate stamps every
        // member's manifest with the instrument.
        assert_eq!(
            manifest["instrument"].as_str(),
            Some("GER40"),
            "the dissolved sweep stamps the instrument: {line}"
        );
        assert!(
            params.iter().any(|p| p[0].as_str() == Some("stop_length"))
                && params.iter().any(|p| p[0].as_str() == Some("stop_k")),
            "the dissolved sweep stamps the resolved stop: {line}"
        );
    }

    // exactly one new Sweep family persisted, with the grid's four members.
    let fams = std::process::Command::new(BIN)
        .args(["runs", "families"])
        .current_dir(&dir)
        .output()
        .unwrap();
    assert!(fams.status.success(), "families exit: {:?}", fams.status);
    let fams_out = String::from_utf8_lossy(&fams.stdout).into_owned();
    assert_eq!(fams_out.lines().count(), 1, "exactly one family persisted: {fams_out}");
    assert!(fams_out.contains("\"kind\":\"Sweep\""), "families: {fams_out}");
    assert!(fams_out.contains("\"members\":4"), "2x2 grid -> four members: {fams_out}");

    // Auto-registration: generated documents are durable intent — exactly
    // one process doc, one campaign doc, one campaign run record.
    let count = |sub: &str| {
        std::fs::read_dir(dir.join("runs").join(sub))
            .map(|d| d.count())
            .unwrap_or(0)
    };
    assert_eq!(count("processes"), 1, "one generated process document registered");
    assert_eq!(count("campaigns"), 1, "one generated campaign document registered");
    let runs_log = std::fs::read_to_string(dir.join("runs").join("campaign_runs.jsonl"))
        .expect("campaign_runs.jsonl exists after a sugar run");
    assert_eq!(runs_log.lines().count(), 1, "one campaign run recorded");
    // `CampaignRunRecord` (aura-registry's lineage.rs)
    // carries no name field at all (campaign id/process id/run/seed/cells/
    // generalizations/trace_name only) — the `--name` handle is NOT part of
    // the run-log line. It IS part of the generated CAMPAIGN DOCUMENT
    // (`translate_sweep` sets `campaign.name = name`), so the handle is
    // checked on that persisted document instead.
    let campaigns_dir = dir.join("runs").join("campaigns");
    let campaign_doc_path = std::fs::read_dir(&campaigns_dir)
        .expect("campaigns dir exists")
        .next()
        .expect("exactly one campaign document")
        .expect("readable dir entry")
        .path();
    let campaign_doc_json = std::fs::read_to_string(&campaign_doc_path).expect("read campaign doc");
    assert!(
        campaign_doc_json.contains("\"name\":\"brp\""),
        "the --name handle lands on the generated campaign document: {campaign_doc_json}"
    );

    // Property (dispatch-rewire seam, docs/specs/sweep-dissolution.md Task 4
    // Step 3): the CLI's `--axis` names are typed against the WRAPPED probe
    // namespace (`sma_signal.fast.length`), but a campaign document's
    // `strategies[].axes` speak the RAW `param_space` namespace
    // (`fast.length`) — `bind_axes`'s convention (#203). `dispatch_sweep`
    // strips the wrapper's one leading node segment before generating the
    // document; a regression that stopped stripping (or stripped one segment
    // too many/few) would persist the wrong axis keys and
    // `validate_campaign_refs` would refuse every axis as unknown before any
    // member ran — this pin catches that at the artifact level, directly.
    let campaign_doc: serde_json::Value =
        serde_json::from_str(&campaign_doc_json).expect("generated campaign doc parses as JSON");
    let axes = campaign_doc["strategies"][0]["axes"]
        .as_object()
        .expect("strategies[0].axes is an object");
    assert!(
        axes.contains_key("fast.length"),
        "the raw (unwrapped) axis name is the document key: {campaign_doc_json}"
    );
    assert!(
        axes.contains_key("slow.length"),
        "the raw (unwrapped) axis name is the document key: {campaign_doc_json}"
    );
    assert!(
        !axes.contains_key("sma_signal.fast.length") && !axes.contains_key("sma_signal.slow.length"),
        "the wrapped CLI probe name must never leak into the stored document: {campaign_doc_json}"
    );

    // Property (dispatch-rewire seam: the ms<->ns crossing, C3): the probed
    // archive window is carried through `DataSource::full_window` (aura's
    // native epoch-ns `Timestamp`, clamped to the archive's actual first/last
    // bar inside the requested range) and converted back to Unix-ms via
    // `aura_ingest::epoch_ns_to_unix_ms` exactly once before landing in the
    // generated document's `data.windows[0]`. A regression that skipped the
    // conversion (or applied it twice, or swapped ms<->ns) would persist a
    // window off by a factor of ~1e6 from the requested `--from`/`--to` —
    // silently, since the executor would still run (over the wrong or an
    // empty range) rather than refuse outright. Pin the sane invariant that
    // survives the archive's exact clamp: the stored window is a
    // (from <= to) sub-range of the requested Unix-ms bounds, not a value
    // orders of magnitude away.
    let from_req = GER40_SEPT2024_FROM_MS.parse::<i64>().unwrap();
    let to_req = GER40_SEPT2024_TO_MS.parse::<i64>().unwrap();
    let window = &campaign_doc["data"]["windows"][0];
    let from_stored = window["from_ms"].as_i64().expect("from_ms is an integer");
    let to_stored = window["to_ms"].as_i64().expect("to_ms is an integer");
    assert!(
        from_req <= from_stored && from_stored <= to_stored && to_stored <= to_req,
        "the stored window is a ms sub-range of the requested [{from_req}, {to_req}]: {campaign_doc_json}"
    );

    // Dedup: a second identical invocation reuses both documents (content-
    // addressed) and appends a second run record.
    let out2 = std::process::Command::new(BIN)
        .args([
            "sweep", &fixture,
            "--real", "GER40",
            "--from", GER40_SEPT2024_FROM_MS,
            "--to", GER40_SEPT2024_TO_MS,
            "--axis", "sma_signal.fast.length=2,4",
            "--axis", "sma_signal.slow.length=8,16",
            "--name", "brp",
        ])
        .current_dir(&dir)
        .output()
        .unwrap();
    assert!(out2.status.success(), "second run stderr: {}", String::from_utf8_lossy(&out2.stderr));
    // `family_id` carries a per-execution uniqueness
    // suffix ("the '-{run}' suffix is appended by the registry",
    // aura-campaign/src/exec.rs `run_cell`) — a fresh family record is
    // written on every execution even when the campaign/process documents
    // themselves dedupe, so two identical invocations are NOT literally
    // byte-identical on stdout: only the trailing run-suffix (0 -> 1)
    // differs. C1 determinism is checked at the member-report level instead
    // (every report field byte-identical), plus the run-suffix increment.
    let out2_stdout = String::from_utf8_lossy(&out2.stdout).into_owned();
    let lines1: Vec<&str> = stdout.lines().collect();
    let lines2: Vec<&str> = out2_stdout.lines().collect();
    assert_eq!(lines2.len(), lines1.len(), "same member count on the dedup run: {out2_stdout}");
    for (l1, l2) in lines1.iter().zip(&lines2) {
        let v1: serde_json::Value = serde_json::from_str(l1).expect("first run member line parses");
        let v2: serde_json::Value = serde_json::from_str(l2).expect("second run member line parses");
        assert_eq!(v1["report"], v2["report"], "identical invocations reproduce identical member reports (C1): {l1} vs {l2}");
        let (fam1, fam2) = (v1["family_id"].as_str().unwrap(), v2["family_id"].as_str().unwrap());
        assert_eq!(
            fam1.trim_end_matches(|c: char| c.is_ascii_digit()),
            fam2.trim_end_matches(|c: char| c.is_ascii_digit()),
            "same family name up to the run-suffix: {fam1} vs {fam2}"
        );
        assert!(fam1.ends_with("-0") && fam2.ends_with("-1"), "run-suffix increments 0 -> 1: {fam1} vs {fam2}");
    }
    assert_eq!(count("processes"), 1, "second identical run dedupes the process doc");
    assert_eq!(count("campaigns"), 1, "second identical run dedupes the campaign doc");
}

/// True iff `root` contains, at any depth, a regular file of non-zero length —
/// the layout-agnostic "traces actually landed on disk" probe. #224 does not yet
/// pin the per-member trace-directory naming (the sweep sugar's cell has no
/// nominee and `persist_campaign_traces` keys by `<trace_name>/<cell_key>`), so
/// this walks the tree rather than hard-coding a path, keeping the headline about
/// *that* traces are written, not *where*.
fn any_nonempty_file(root: &Path) -> bool {
    let mut stack = vec![root.to_path_buf()];
    while let Some(p) = stack.pop() {
        let Ok(entries) = std::fs::read_dir(&p) else { continue };
        for entry in entries.flatten() {
            let path = entry.path();
            if path.is_dir() {
                stack.push(path);
            } else if std::fs::metadata(&path).map(|m| m.len() > 0).unwrap_or(false) {
                return true;
            }
        }
    }
    false
}

/// Property (#224): `aura sweep <blueprint.json> --real SYM --axis … --trace <fam>`
/// DELIVERS the advertised trace-writing capability instead of the #168
/// refuse-with-pointer. Two things must hold once the capability ships:
/// (1) the run no longer refuses (`aura: --trace is not yet available on sweep;
/// see #224`, exit 2) but succeeds, and (2) the tap series actually land on disk
/// under `runs/traces/` — at least one non-empty tap file is written, the honest
/// inverse of the #168 no-op where the `--real` sugar set `persist_taps: vec![]`
/// (verb_sugar.rs) and `run_blueprint_sweep` did `let _ = persist`, so nothing was
/// ever written. The exact on-disk layout (trace name + per-member/nominee key)
/// is deliberately NOT pinned here: this headline pins that traces are written at
/// all — not how many dirs, nor under what member key — because that naming is
/// the GREEN slice's open design point (the selection-free sweep cell has no
/// nominee for the existing per-cell `persist_campaign_traces` mechanism, #224).
/// Gated on the local GER40 archive (the project skip-on-no-data convention);
/// the refused status quo makes it RED on a data-ful host today.
#[test]
fn sweep_real_with_trace_writes_tap_series_to_disk() {
    if !local_data_present() {
        eprintln!("skip: no local data at {}", data_server::DEFAULT_DATA_PATH);
        return;
    }
    let (dir, _g) = fresh_project();
    let fixture = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let out = std::process::Command::new(BIN)
        .args([
            "sweep", &fixture,
            "--real", "GER40",
            "--from", GER40_SEPT2024_FROM_MS,
            "--to", GER40_SEPT2024_TO_MS,
            "--axis", "sma_signal.fast.length=2,4",
            "--axis", "sma_signal.slow.length=8",
            "--trace", "brp",
        ])
        .current_dir(&dir)
        .output()
        .unwrap();
    // (1) the #168 refusal is lifted: the advertised flag runs instead of exit 2.
    assert!(
        out.status.success(),
        "sweep --trace must no longer refuse (#224); exit {:?}, stderr: {}",
        out.status.code(),
        String::from_utf8_lossy(&out.stderr),
    );
    // (2) the tap series actually land on disk under runs/traces/ (non-empty) —
    //     the honest inverse of the #168 `persist_taps: vec![]` no-op.
    let traces_root = dir.join("runs").join("traces");
    assert!(
        any_nonempty_file(&traces_root),
        "sweep --trace must write at least one non-empty tap file under {} \
         (found none — the #168 no-op wrote nothing); stderr: {}",
        traces_root.display(),
        String::from_utf8_lossy(&out.stderr),
    );
}

/// Extract the first persisted family whose `kind` matches `want` from
/// `aura runs families` stdout (one JSON object per line), returning its
/// `family_id` — the id `aura reproduce` addresses. Panics if none matches so a
/// mis-minted store fails loudly rather than skipping the assertion.
fn family_id_of_kind(dir: &Path, want: &str) -> String {
    let out = Command::new(BIN).args(["runs", "families"]).current_dir(dir).output().unwrap();
    assert!(out.status.success(), "runs families exit: {:?}", out.status);
    let stdout = String::from_utf8_lossy(&out.stdout).into_owned();
    for line in stdout.lines() {
        let v: serde_json::Value = serde_json::from_str(line).expect("family line parses as JSON");
        if v["kind"].as_str() == Some(want) {
            return v["family_id"].as_str().expect("family_id is a string").to_string();
        }
    }
    panic!("no {want} family in the store: {stdout}");
}

/// Property (C18/C1 reproduce guarantee): `aura reproduce` re-derives a family
/// over the SAME data it was minted on. A sweep family minted over real GER40
/// bars must therefore reproduce bit-identically — the run is deterministic
/// (a fresh identical sweep matches the persisted metrics to the last digit), so
/// the README's "check it is bit-identical" promise must hold. It must NOT
/// silently re-run every member over the built-in synthetic showcase stream,
/// which diverges from the stored real-data metrics on every member (observed:
/// `reproduced 0/N bit-identically`, exit 1). Gated on the local GER40 archive
/// (skip-on-no-data convention); uses the plain `r_sma.json` blueprint —
/// no gang, no campaign-specific fixture — because the divergence is driven by
/// `--real`, not by the minting path.
#[test]
fn reproduce_real_sweep_family_re_derives_bit_identically() {
    if !local_data_present() {
        eprintln!("skip: no local data at {}", data_server::DEFAULT_DATA_PATH);
        return;
    }
    let (dir, _g) = fresh_project();
    let fixture = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let mint = Command::new(BIN)
        .args([
            "sweep", &fixture,
            "--real", "GER40",
            "--from", GER40_SEPT2024_FROM_MS,
            "--to", GER40_SEPT2024_TO_MS,
            "--axis", "sma_signal.fast.length=2,4",
            "--axis", "sma_signal.slow.length=8",
            "--name", "repro_sweep",
        ])
        .current_dir(&dir)
        .output()
        .unwrap();
    assert!(mint.status.success(), "sweep mint stderr: {}", String::from_utf8_lossy(&mint.stderr));

    let id = family_id_of_kind(&dir, "Sweep");
    let out = Command::new(BIN).args(["reproduce", &id]).current_dir(&dir).output().unwrap();
    let stdout = String::from_utf8_lossy(&out.stdout).into_owned();
    assert!(
        out.status.success(),
        "reproduce of a real-data sweep family must succeed (exit 0), got {:?}\nstdout: {stdout}\nstderr: {}",
        out.status.code(),
        String::from_utf8_lossy(&out.stderr)
    );
    assert!(
        stdout.contains("reproduced 2/2 members bit-identically"),
        "every member of a deterministic real-data sweep re-derives bit-identically: {stdout}"
    );
    assert!(
        !stdout.contains("DIVERGED"),
        "no member may DIVERGE — the run is deterministic, reproduce must use the recorded data source: {stdout}"
    );
}

/// Property (C18/C1 reproduce guarantee, #246): the same reproduce guarantee holds
/// when the swept axis names a BOUND param of a CLOSED starter blueprint, not an
/// already-open one — `r_sma.json` (both `sma_signal.fast.length` and
/// `sma_signal.slow.length` bound), overriding only the `fast.length` axis and
/// leaving `slow.length` bound — unlike the walk-forward reproduce sibling
/// (`aura_walkforward_over_a_blueprint_reproduces_bit_identically`), which
/// overrides both axes over the same closed blueprint.
/// Modelled on that sibling (synthetic in-process walk-forward — no `--real`),
/// NOT on `reproduce_real_sweep_family_re_derives_bit_identically`: a real-data
/// mint routes through the campaign executor (`CliMemberRunner::run_member` +
/// `validate_and_register_axes`), neither of which carries any #246 override
/// threading — a gap outside this task's file scope (`blueprint_sweep_over`,
/// `reproduce_family_in`) and spanning all four dispatch verbs, so exercising
/// it here would silently expand into unrelated, unreviewed surface. The
/// synthetic walk-forward path exercises exactly the threaded functions:
/// `blueprint_sweep_over` (via `blueprint_walkforward_family`) re-opens
/// `sma_signal.fast.length` for the per-window IS re-fit, and
/// `reproduce_family_in` re-opens it again from the recorded manifest params
/// so the re-derivation resolves against the same space the mint used.
#[test]
fn reproduce_family_with_overridden_bound_param_re_derives() {
    let cwd = temp_cwd("walkforward-reproduce-override");
    let bp = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let run = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args(["walkforward", &bp, "--axis", "sma_signal.fast.length=2,3", "--name", "wfo"])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura walkforward");
    assert_eq!(
        run.status.code(),
        Some(0),
        "walk-forward over a bound-param axis must succeed (#246 re-open), stderr: {}",
        String::from_utf8_lossy(&run.stderr)
    );
    let repro = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args(["reproduce", "wfo-0"])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura reproduce");
    assert_eq!(
        repro.status.code(),
        Some(0),
        "reproduce stderr: {}",
        String::from_utf8_lossy(&repro.stderr)
    );
    let out = String::from_utf8(repro.stdout).expect("utf-8");
    assert!(
        out.contains("reproduced 3/3 members bit-identically"),
        "every OOS member of a deterministic walk-forward over a bound-param axis \
         re-derives bit-identically: {out}"
    );
    let _ = std::fs::remove_dir_all(&cwd);
}

/// Property (#246, distinct code path from `reproduce_family_with_overridden_bound_param_re_derives`):
/// `reproduce_family_in`'s override derivation (`wrapped_bound_overrides_of`, computed once
/// per family from the FIRST member's recorded params) applies uniformly across
/// `FamilyKind`s, not only `WalkForward` — a plain `aura sweep` family minted by
/// overriding a bound param on the CLOSED `r_sma.json` fixture re-derives
/// bit-identically too. The sibling test exercises `FamilyKind::WalkForward`'s
/// branch (per-window OOS reload via `run_oos_blueprint`); this one exercises the
/// `Sweep`-kind branch (`data.run_sources`/`data.full_window`, no per-window
/// re-fit) of the same shared dispatch in `reproduce_family_in`. A regression that
/// threaded the #246 override set through one branch but not the other would pass
/// the walk-forward sibling while failing here.
#[test]
fn aura_reproduce_re_derives_a_sweep_family_minted_over_a_bound_param_override() {
    let cwd = temp_cwd("sweep-reproduce-bound-override");
    let bp = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let sweep = Command::new(BIN)
        .args(["sweep", &bp, "--axis", "sma_signal.fast.length=2,3", "--name", "swo"])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura sweep over a bound-param axis");
    assert_eq!(
        sweep.status.code(),
        Some(0),
        "sweeping a bound param of a closed blueprint must succeed (#246 re-open), stderr: {}",
        String::from_utf8_lossy(&sweep.stderr)
    );
    let repro = Command::new(BIN)
        .args(["reproduce", "swo-0"])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura reproduce");
    assert_eq!(
        repro.status.code(),
        Some(0),
        "reproduce stderr: {}",
        String::from_utf8_lossy(&repro.stderr)
    );
    let out = String::from_utf8(repro.stdout).expect("utf-8");
    assert!(
        out.contains("reproduced 2/2 members bit-identically"),
        "every member of a plain sweep minted over a bound-param override \
         re-derives bit-identically: {out}"
    );
    let _ = std::fs::remove_dir_all(&cwd);
}

/// Property: `aura reproduce` yields a handled outcome — a reproduction verdict
/// or a clean `aura:` refusal — for a real-data WalkForward family, and NEVER an
/// internal panic. A WalkForward member's OOS window is stored as real epoch-ns
/// bounds; reconstructing that window over the built-in synthetic walk (whose 60
/// bars fall entirely outside those bounds) yields an empty source, which today
/// reaches the unguarded `window_of(&s).expect("non-empty OOS window")` and
/// aborts the process (exit 101). Gated on the local GER40 archive
/// (skip-on-no-data convention); a ~135-day window is the minimum that stitches
/// at least one OOS window (IS 90d + OOS 30d).
#[test]
fn reproduce_real_walkforward_family_does_not_panic() {
    if !local_data_present() {
        eprintln!("skip: no local data at {}", data_server::DEFAULT_DATA_PATH);
        return;
    }
    // 2025-01-01 .. ~2025-05-16 (~135 days) — long enough for one IS(90d)+OOS(30d)
    // walk-forward window over the local GER40 archive.
    const WF_FROM_MS: &str = "1735689600000";
    const WF_TO_MS: &str = "1747353600000";
    let (dir, _g) = fresh_project();
    let fixture = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let mint = Command::new(BIN)
        .args([
            "walkforward", &fixture,
            "--real", "GER40",
            "--from", WF_FROM_MS,
            "--to", WF_TO_MS,
            "--axis", "sma_signal.fast.length=2,4",
            "--axis", "sma_signal.slow.length=8",
            "--name", "repro_wf",
        ])
        .current_dir(&dir)
        .output()
        .unwrap();
    assert!(mint.status.success(), "walkforward mint stderr: {}", String::from_utf8_lossy(&mint.stderr));

    let id = family_id_of_kind(&dir, "WalkForward");
    let out = Command::new(BIN).args(["reproduce", &id]).current_dir(&dir).output().unwrap();
    let stdout = String::from_utf8_lossy(&out.stdout).into_owned();
    let stderr = String::from_utf8_lossy(&out.stderr).into_owned();
    assert_ne!(
        out.status.code(),
        Some(101),
        "reproduce of a WalkForward family must not panic (exit 101)\nstdout: {stdout}\nstderr: {stderr}"
    );
    assert!(
        !stderr.contains("panicked") && !stderr.contains("non-empty OOS window"),
        "no internal panic may leak from reproduce: {stderr}"
    );
    assert!(
        stdout.contains("reproduced") || stderr.starts_with("aura:"),
        "reproduce must yield a verdict (stdout) or a clean `aura:` refusal (stderr): stdout={stdout} stderr={stderr}"
    );
}

// Fixture blueprints for the verb-level multi-column binding coverage below
// (#231 task 4 quality follow-up): a CLOSED high/low blueprint (`run`
// requires zero free knobs) and an OPEN high/low blueprint with two
// SMA-length axes (`walkforward` requires >= 1 axis). Written to a fresh
// temp file per test rather than a tracked `examples/` fixture — these exist
// only to drive the refusal / real-open contract, not as authoring samples.
const HL_RANGE_CLOSED_BLUEPRINT: &str = r#"{
  "format_version": 1,
  "blueprint": {
    "name": "hl_range",
    "nodes": [ {"primitive":{"type":"Sub"}} ],
    "edges": [],
    "input_roles": [
      {"name":"high","targets":[{"node":0,"slot":0}],"source":"F64"},
      {"name":"low","targets":[{"node":0,"slot":1}],"source":"F64"}
    ],
    "output": [{"node":0,"field":0,"name":"bias"}]
  }
}"#;
const HL_SIGNAL_OPEN_BLUEPRINT: &str = r#"{"format_version":1,"blueprint":{"name":"hl_signal","nodes":[{"primitive":{"type":"SMA","name":"fast"}},{"primitive":{"type":"SMA","name":"slow"}},{"primitive":{"type":"Sub"}},{"primitive":{"type":"Bias","name":"bias","bound":[{"pos":0,"name":"scale","kind":"F64","value":{"F64":0.5}}]}}],"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}],"input_roles":[{"name":"high","targets":[{"node":0,"slot":0}],"source":"F64"},{"name":"low","targets":[{"node":1,"slot":0}],"source":"F64"}],"output":[{"node":3,"field":0,"name":"bias"}]}}"#;

/// Property (#231 task 4 quality follow-up): `aura run` over a multi-column
/// blueprint with NO `--real` (the synthetic default) refuses through the
/// verb-level exit-1 + stderr contract — `binding::synthetic_refusal`'s exact
/// prose, not merely its `Result` shape (the unit-mod test one layer down
/// covers the `Err`-returning builders only). Data-free: the guard fires
/// directly after binding resolution, before any archive access, so no
/// gating is needed.
#[test]
fn run_synthetic_refuses_a_multi_column_blueprint_before_data_access() {
    let dir = temp_cwd("run_synthetic_multicolumn_refusal");
    let fixture = dir.join("hl_range.json");
    std::fs::write(&fixture, HL_RANGE_CLOSED_BLUEPRINT).expect("write fixture");
    let out = Command::new(BIN).args(["run", fixture.to_str().unwrap()]).output().expect("spawn aura run");
    assert_eq!(out.status.code(), Some(1), "status: {:?}", out.status);
    let stderr = String::from_utf8_lossy(&out.stderr);
    assert_eq!(
        stderr.trim_end(),
        "aura: strategy \"hl_range\" consumes columns beyond close (high, low) — synthetic \
         data generates a close series only; run with --real <SYMBOL>"
    );
    let _ = std::fs::remove_dir_all(&dir);
}

/// Property (#231 task 4 quality follow-up): the sibling exit-process arm at
/// `blueprint_walkforward_family` — `aura walkforward` over a multi-column
/// blueprint with no `--real` — refuses identically. Data-free (the guard
/// fires directly after binding resolution, before any window/archive
/// access); run inside a temp cwd purely by the project's verb-test
/// convention (no store write is expected on this refusal path either).
#[test]
fn walkforward_synthetic_refuses_a_multi_column_blueprint_before_data_access() {
    let dir = temp_cwd("walkforward_synthetic_multicolumn_refusal");
    let fixture = dir.join("hl_signal.json");
    std::fs::write(&fixture, HL_SIGNAL_OPEN_BLUEPRINT).expect("write fixture");
    let out = Command::new(BIN)
        .args([
            "walkforward", fixture.to_str().unwrap(),
            "--axis", "hl_signal.fast.length=2,4",
            "--axis", "hl_signal.slow.length=8",
        ])
        .current_dir(&dir)
        .output()
        .expect("spawn aura walkforward");
    assert_eq!(out.status.code(), Some(1), "status: {:?}", out.status);
    let stderr = String::from_utf8_lossy(&out.stderr);
    assert_eq!(
        stderr.trim_end(),
        "aura: strategy \"hl_signal\" consumes columns beyond close (high, low) — synthetic \
         data generates a close series only; run with --real <SYMBOL>"
    );
    assert!(!dir.join("runs").exists(), "a refused synthetic family must not start a run store");
    let _ = std::fs::remove_dir_all(&dir);
}

/// Property (#231 task 4 quality follow-up): the positive counterpart of the
/// two refusals above — a multi-column blueprint over `--real GER40` opens
/// BOTH the high and low columns end to end (`open_real_source` ->
/// `resolve_run_data` -> `run_signal_r`) and produces a real `RunReport`,
/// never the single-close weld the pre-task-4 code hardcoded. Gated on the
/// local GER40 archive (skip-on-no-data convention).
#[test]
fn run_real_multi_column_blueprint_opens_high_and_low_columns() {
    if !local_data_present() {
        eprintln!("skip: no local data at {}", data_server::DEFAULT_DATA_PATH);
        return;
    }
    let dir = temp_cwd("run_real_multicolumn_open");
    let fixture = dir.join("hl_range.json");
    std::fs::write(&fixture, HL_RANGE_CLOSED_BLUEPRINT).expect("write fixture");
    let out = Command::new(BIN)
        .args([
            "run", fixture.to_str().unwrap(),
            "--real", "GER40",
            "--from", GER40_SEPT2024_FROM_MS,
            "--to", GER40_SEPT2024_TO_MS,
        ])
        .output()
        .expect("spawn aura run");
    assert_eq!(
        out.status.code(), Some(0),
        "a high/low blueprint over real GER40 data must run cleanly, got status={:?} stderr={}",
        out.status, String::from_utf8_lossy(&out.stderr)
    );
    let stdout = String::from_utf8_lossy(&out.stdout);
    assert!(
        stdout.trim_start().starts_with("{\"manifest\":"),
        "exit 0 must carry a JSON report, got: {stdout}"
    );
    let _ = std::fs::remove_dir_all(&dir);
}

/// Property (#231 task 4, "every real-data path" — `run --real` above covers
/// only `open_real_source`/`run_signal_r`; `sweep --real` walks a DISTINCT
/// real-data path, `blueprint_sweep_family` -> `blueprint_sweep_over` ->
/// `DataSource::windowed_sources`, which independently needed `binding.
/// columns()` threaded through in task 4): a multi-column (high/low) OPEN
/// blueprint swept over `--real GER40` opens both columns per member and
/// produces one member line per grid point, not a source-count-mismatch panic
/// or a fallback to a single close source. Gated on the local GER40 archive.
#[test]
fn sweep_real_multi_column_blueprint_opens_high_and_low_columns() {
    if !local_data_present() {
        eprintln!("skip: no local data at {}", data_server::DEFAULT_DATA_PATH);
        return;
    }
    let (dir, _g) = fresh_project();
    let fixture_dir = temp_cwd("sweep_real_multicolumn_open");
    let fixture = fixture_dir.join("hl_signal.json");
    std::fs::write(&fixture, HL_SIGNAL_OPEN_BLUEPRINT).expect("write fixture");
    let out = std::process::Command::new(BIN)
        .args([
            "sweep", fixture.to_str().unwrap(),
            "--real", "GER40",
            "--from", GER40_SEPT2024_FROM_MS,
            "--to", GER40_SEPT2024_TO_MS,
            "--axis", "hl_signal.fast.length=2,4",
            "--axis", "hl_signal.slow.length=8",
        ])
        .current_dir(&dir)
        .output()
        .unwrap();
    assert!(out.status.success(), "stderr: {}", String::from_utf8_lossy(&out.stderr));
    let stdout = String::from_utf8_lossy(&out.stdout).into_owned();
    let lines: Vec<&str> = stdout.lines().collect();
    assert_eq!(lines.len(), 2, "one member line per grid point: {stdout}");
    for line in &lines {
        let v: serde_json::Value = serde_json::from_str(line).expect("member line parses as JSON");
        assert!(
            v["report"]["manifest"].is_object(),
            "each member carries a real report, not a source-mismatch fault: {line}"
        );
    }
    let _ = std::fs::remove_dir_all(&fixture_dir);
}

/// Property (#210 c0110 fieldtest, "the wrapped-name refusal mangles the axis
/// the user typed"): the dissolved real-data blueprint sweep types `--axis`
/// against the WRAPPED probe namespace (`sma_signal.fast.length`, what
/// `--list-axes` prints), not the raw campaign-document namespace
/// (`fast.length`). A user who types the raw form must be refused up front,
/// echoing exactly what they typed — never a stripped-then-mangled fragment
/// (the old failure mode: `fast.length` -> strip one segment -> the
/// nonsensical bare `"length"`). NOT gated: the preflight is data-free (it
/// only reloads the blueprint to probe its param space), so the refusal
/// fires before the archive is ever touched — this test needs no local data.
#[test]
fn aura_sweep_real_blueprint_refuses_a_raw_form_axis_name_before_data_access() {
    let dir = temp_cwd("sweep_real_rawname_refusal");
    let fixture = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let out = Command::new(BIN)
        .args([
            "sweep", &fixture,
            "--real", "GER40",
            "--from", GER40_SEPT2024_FROM_MS,
            "--to", GER40_SEPT2024_TO_MS,
            "--axis", "fast.length=2,4",
            "--name", "c0110-rawname",
        ])
        .current_dir(&dir)
        .output()
        .expect("spawn aura sweep (raw-form axis)");
    assert_eq!(out.status.code(), Some(2), "a raw-form axis name must be refused, not run: status={:?}", out.status);
    let stderr = String::from_utf8_lossy(&out.stderr).into_owned();
    assert!(
        stderr.contains("axis \"fast.length\""),
        "the refusal echoes exactly the name the user typed: {stderr}"
    );
    assert!(
        !stderr.contains("axis \"length\""),
        "must never mangle the typed axis into a stripped fragment: {stderr}"
    );
    // Data-free: no archive/store access at all.
    assert!(!dir.join("runs").exists(), "a refused axis name must not start a real run");
    let _ = std::fs::remove_dir_all(&dir);
}

/// Property (#247): every sweep-verb boundary `BindError` renders as prose in
/// the same register as its unknown-axis sibling
/// (`aura: axis <name>: names no param of this blueprint … — see `aura sweep
/// <bp> --list-axes``) — never a raw Rust Debug struct. A wrong-kind axis value
/// and an uncovered open knob are the two variants that today leak
/// `KindMismatch { knob: …, expected: …, got: … }` and
/// `MissingKnob("…")` verbatim to stderr; both must instead name the axis, carry
/// the specifics (expected vs supplied kind / the unbound knob), and point at
/// `--list-axes`, with the exit code staying 2. Data-free: both faults are
/// caught at the axis-resolution boundary before any member runs, so the test
/// needs no archive and no project (hostless synthetic sweep).
#[test]
fn sweep_boundary_bind_errors_render_as_prose_not_debug_structs() {
    let dir = temp_cwd("sweep-bind-error-prose");

    // (a) KindMismatch — an axis re-opens the bound I64 `fast.length` (#246) but
    // supplies F64 values: the specifics are the expected vs supplied kind.
    let closed = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let km = Command::new(BIN)
        .args(["sweep", &closed, "--axis", "sma_signal.fast.length=2.5,3.5"])
        .current_dir(&dir)
        .output()
        .expect("spawn aura sweep (kind-mismatched axis)");
    assert_eq!(
        km.status.code(),
        Some(2),
        "a kind-mismatched axis value stays a usage refusal: status={:?}",
        km.status
    );
    let km_err = String::from_utf8_lossy(&km.stderr).into_owned();
    assert!(
        !km_err.contains("KindMismatch"),
        "must not leak the raw Rust Debug struct name: {km_err}"
    );
    assert!(
        km_err.contains("sma_signal.fast.length"),
        "the prose names the offending axis exactly as typed: {km_err}"
    );
    assert!(
        km_err.contains("I64") && km_err.contains("F64"),
        "the prose carries the expected vs supplied kind specifics: {km_err}"
    );
    assert!(
        km_err.contains("--list-axes"),
        "the prose points at the discovery surface, like the unknown-axis sibling: {km_err}"
    );

    // (b) MissingKnob — the open fixture leaves `slow.length` unbound when only
    // `fast.length` is swept: the specific is the uncovered open knob.
    let open = format!("{}/tests/fixtures/r_sma_open.json", env!("CARGO_MANIFEST_DIR"));
    let mk = Command::new(BIN)
        .args(["sweep", &open, "--axis", "sma_signal.fast.length=2,4"])
        .current_dir(&dir)
        .output()
        .expect("spawn aura sweep (uncovered open knob)");
    assert_eq!(
        mk.status.code(),
        Some(2),
        "an uncovered open knob stays a usage refusal: status={:?}",
        mk.status
    );
    let mk_err = String::from_utf8_lossy(&mk.stderr).into_owned();
    assert!(
        !mk_err.contains("MissingKnob"),
        "must not leak the raw Rust Debug struct name: {mk_err}"
    );
    assert!(
        mk_err.contains("sma_signal.slow.length"),
        "the prose names the unbound knob no axis supplies: {mk_err}"
    );
    assert!(
        mk_err.contains("--list-axes"),
        "the prose points at the discovery surface, like the unknown-axis sibling: {mk_err}"
    );

    let _ = std::fs::remove_dir_all(&dir);
}

/// Property (#210 c0110 fieldtest, "refused dissolved sweeps still register
/// their generated campaign document" — store litter, probe P3): binding only
/// a SUBSET of the blueprint's open axes (here: `fast.length`, leaving
/// `slow.length` unbound) is a referential refusal at member-run time — but
/// the sugar must validate the generated documents (including the
/// axis-binding shape) BEFORE `register_generated`, so the refusal leaves
/// `runs/processes/` and `runs/campaigns/` untouched and appends no
/// `campaign_runs.jsonl` record. Gated on the local GER40 archive (the
/// referential refusal is reached only after the archive is probed for the
/// window, same as the sibling dissolved-sweep pin above).
#[test]
fn aura_sweep_real_blueprint_subset_axes_refuses_without_store_litter() {
    if !local_data_present() {
        eprintln!("skip: no local data at {}", data_server::DEFAULT_DATA_PATH);
        return;
    }
    let (dir, _g) = fresh_project();
    let fixture = format!("{}/tests/fixtures/r_sma_open.json", env!("CARGO_MANIFEST_DIR"));
    let out = Command::new(BIN)
        .args([
            "sweep", &fixture,
            "--real", "GER40",
            "--from", GER40_SEPT2024_FROM_MS,
            "--to", GER40_SEPT2024_TO_MS,
            "--axis", "sma_signal.fast.length=2,4",
            "--name", "c0110-subset",
        ])
        .current_dir(&dir)
        .output()
        .expect("spawn aura sweep (subset axes)");
    assert_ne!(
        out.status.code(),
        Some(0),
        "leaving slow.length unbound must refuse, not run: stdout={}",
        String::from_utf8_lossy(&out.stdout)
    );
    let count = |sub: &str| {
        std::fs::read_dir(dir.join("runs").join(sub)).map(|d| d.count()).unwrap_or(0)
    };
    assert_eq!(count("processes"), 0, "a refused sweep must not register a generated process document");
    assert_eq!(count("campaigns"), 0, "a refused sweep must not register a generated campaign document");
    let runs_log = dir.join("runs").join("campaign_runs.jsonl");
    assert!(
        !runs_log.exists() || std::fs::read_to_string(&runs_log).unwrap().is_empty(),
        "no realization recorded for a refused sweep"
    );
}

/// Property (issue #246, task 5): the dissolved real-data `aura sweep`
/// campaign path accepts an `--axis` naming a BOUND param of a CLOSED
/// blueprint, not only an already-open one. Tasks 3-4 threaded this override
/// convention (bound value = default, `Composite::reopen`) through the
/// SYNTHETIC blueprint verb paths (`blueprint_sweep_family`/
/// `blueprint_sweep_over`, `reproduce_family_with_overridden_bound_param_re_derives`);
/// the real-data dissolved path routes through a distinct seam
/// (`CliMemberRunner::run_member` + `verb_sugar::validate_before_register`'s
/// P3 preflight) that carried no such threading at all, the explicit gap
/// named on `reproduce_family_with_overridden_bound_param_re_derives`'s doc
/// comment. Modelled on `sweep_real_blueprint_member_lines_pin_the_dissolved_contract`
/// (the dissolved-sweep e2e family `walkforward_dissolved_refuses_an_unknown_axis_name`
/// belongs to): same store-reading assertions (`aura runs families`, one
/// persisted `Sweep` family, member count), over the same `r_sma.json`
/// fixture every other real-data sweep pin uses (`sma_signal.fast.length`/
/// `sma_signal.slow.length` both bound) — the distinguishing property here
/// is that the swept axis names an already-bound param, not that the
/// fixture differs.
#[test]
fn sweep_dissolved_accepts_an_axis_over_a_bound_param() {
    if !local_data_present() {
        eprintln!("skip: no local data at {}", data_server::DEFAULT_DATA_PATH);
        return;
    }
    let (dir, _g) = fresh_project();
    let fixture = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let out = Command::new(BIN)
        .args([
            "sweep", &fixture,
            "--real", "GER40",
            "--from", GER40_SEPT2024_FROM_MS,
            "--to", GER40_SEPT2024_TO_MS,
            "--axis", "sma_signal.fast.length=2,3",
            "--name", "bound-override",
        ])
        .current_dir(&dir)
        .output()
        .expect("spawn aura sweep over a bound-param axis");
    assert_eq!(
        out.status.code(),
        Some(0),
        "a bound-param axis must re-open (#246) on the dissolved real-data campaign \
         path too, stderr: {}",
        String::from_utf8_lossy(&out.stderr)
    );
    let fams = Command::new(BIN).args(["runs", "families"]).current_dir(&dir).output().unwrap();
    assert!(fams.status.success(), "families exit: {:?}", fams.status);
    let fams_out = String::from_utf8_lossy(&fams.stdout).into_owned();
    assert_eq!(fams_out.lines().count(), 1, "exactly one family persisted: {fams_out}");
    assert!(fams_out.contains("\"kind\":\"Sweep\""), "families: {fams_out}");
    assert!(
        fams_out.contains("\"members\":2"),
        "the two-value bound-param axis grid yields two members: {fams_out}"
    );
}

/// Property (#246 override merge, exercised at scale by #248's open-twins
/// migration): sweeping the CLOSED `examples/r_sma.json` (both SMA lengths
/// bound to defaults 2/4) with `--axis` values that override BOTH bound
/// lengths yields BIT-IDENTICAL per-member `total_pips` to sweeping the
/// historical OPEN `tests/fixtures/r_sma_open.json` (both lengths genuinely
/// unbound) over the same axis values. #248's migration replaced the "bind
/// an open param" idiom with "override a bound param" at dozens of call
/// sites across cli_run.rs/research_docs.rs/project_load.rs; if the override
/// merge ever diverged from the historical bind path (e.g. a different
/// position in the rebuilt param vector), every migrated seed/grade site
/// would silently compile a DIFFERENT graph while still exiting 0 — no other
/// test cross-checks the two idioms against each other. NOT gated — a
/// synthetic sweep, no `--real`, so it is CI-safe on every machine.
#[test]
fn sweep_over_a_bound_override_matches_the_historical_open_bind() {
    let closed_dir = temp_cwd("bound-override-matches-open-closed");
    let closed_bp = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let closed_out = Command::new(BIN)
        .args([
            "sweep", &closed_bp,
            "--axis", "sma_signal.fast.length=2,4",
            "--axis", "sma_signal.slow.length=8,16",
        ])
        .current_dir(&closed_dir)
        .output()
        .expect("spawn sweep over the closed blueprint's bound-override axes");
    assert!(
        closed_out.status.success(),
        "closed sweep exit: {:?}; stderr: {}",
        closed_out.status,
        String::from_utf8_lossy(&closed_out.stderr)
    );

    let open_dir = temp_cwd("bound-override-matches-open-open");
    let open_bp = format!("{}/tests/fixtures/r_sma_open.json", env!("CARGO_MANIFEST_DIR"));
    let open_out = Command::new(BIN)
        .args([
            "sweep", &open_bp,
            "--axis", "sma_signal.fast.length=2,4",
            "--axis", "sma_signal.slow.length=8,16",
        ])
        .current_dir(&open_dir)
        .output()
        .expect("spawn sweep over the open blueprint's unbound axes");
    assert!(
        open_out.status.success(),
        "open sweep exit: {:?}; stderr: {}",
        open_out.status,
        String::from_utf8_lossy(&open_out.stderr)
    );

    let pips = |dir: &Path| -> Vec<f64> {
        let rank = Command::new(BIN)
            .args(["runs", "family", "sweep-0", "rank", "total_pips"])
            .current_dir(dir)
            .output()
            .expect("spawn rank");
        assert!(rank.status.success(), "rank exit: {:?}", rank.status);
        let out = String::from_utf8(rank.stdout).expect("utf-8");
        out.lines()
            .map(|l| {
                let key = "\"total_pips\":";
                let start = l.find(key).expect("line has total_pips") + key.len();
                let tail = &l[start..];
                let end = tail.find([',', '}']).expect("token end");
                tail[..end].parse().expect("total_pips is an f64")
            })
            .collect()
    };
    let mut closed_pips = pips(&closed_dir);
    let mut open_pips = pips(&open_dir);
    closed_pips.sort_by(|a, b| a.partial_cmp(b).unwrap());
    open_pips.sort_by(|a, b| a.partial_cmp(b).unwrap());
    assert_eq!(
        closed_pips, open_pips,
        "bound-override and historical open-bind must compute bit-identical \
         member metrics for the same axis grid"
    );

    let _ = std::fs::remove_dir_all(&closed_dir);
    let _ = std::fs::remove_dir_all(&open_dir);
}

/// The synthetic seed-resweep `mc` is undefined over real bars (one realization ->
/// identical members), so a bare `aura mc --real EURUSD` — with no R candidate — is
/// REJECTED (usage on stderr, exit 2) at the binary boundary. The real path is
/// reachable ONLY via the blueprint positional + `--axis` campaign form (the
/// R-bootstrap over the pooled OOS R series,
/// `mc_r_bootstrap_real_e2e_pins_the_exact_current_grade`; spelled
/// `--strategy r-sma` before the #220 argv migration); a `--real` without a
/// blueprint stays a usage error. NOT gated — the refusal precedes any
/// data access, so it is CI-safe on every machine.
#[test]
fn mc_rejects_real_flag_with_usage_exit_2() {
    let dir = temp_cwd("mc_no_real");
    let out = Command::new(BIN)
        .args(["mc", "--real", "EURUSD"])
        .current_dir(&dir)
        .output()
        .expect("spawn aura mc --real");
    assert_eq!(out.status.code(), Some(2), "mc --real without a blueprint is a usage error");
    let stderr = String::from_utf8_lossy(&out.stderr);
    assert!(stderr.contains("Usage: aura mc"), "usage refusal: {stderr}");

    let _ = std::fs::remove_dir_all(&dir);
}

/// Property (window-flags-require-`--real`, at the binary boundary): a family
/// command given a `--from`/`--to` window WITHOUT `--real` is a usage error —
/// there is no synthetic window knob, so `aura sweep --from 100` and
/// `aura walkforward --to 200` exit 2 (usage on stderr), never a silent synthetic
/// run over the whole built-in stream. The pure grammar is unit-tested in
/// `main.rs`; this pins that the `Err` arm actually wires through to `exit(2)` at
/// the dispatch for BOTH family commands. NOT gated — the refusal is pure
/// argument parsing, before any data access.
#[test]
fn family_window_flag_without_real_refuses_with_usage_exit_2() {
    // (subcommand args, the subcommand token the usage message must name)
    for (args, cmd) in [
        (&["sweep", "--from", "100"][..], "sweep"),
        (&["sweep", "--to", "200"][..], "sweep"),
        (&["walkforward", "--from", "100"][..], "walkforward"),
        (&["walkforward", "--to", "200"][..], "walkforward"),
    ] {
        let dir = temp_cwd("family_window_no_real");
        let out = Command::new(BIN)
            .args(args)
            .current_dir(&dir)
            .output()
            .unwrap_or_else(|e| panic!("spawn aura {args:?}: {e}"));
        assert_eq!(out.status.code(), Some(2), "`aura {args:?}` must exit 2; status: {:?}", out.status);
        assert!(out.stdout.is_empty(), "`aura {args:?}` must emit no run on stdout: {:?}", out.stdout);
        // the refusal carries the subcommand's usage, which names `--real` — the
        // flag the window needs and was given without.
        let stderr = String::from_utf8_lossy(&out.stderr);
        assert!(stderr.contains(cmd), "`aura {args:?}` stderr must name `{cmd}`: {stderr:?}");
        assert!(stderr.contains("--real"), "`aura {args:?}` stderr must name `--real`: {stderr:?}");
        let _ = std::fs::remove_dir_all(&dir);
    }
}

/// Property (house style, #179): every hand-rolled usage line reads
/// `Usage: aura <verb> …` — capital `Usage:` plus the program name inside the
/// grammar. `sweep`/`walkforward`'s built-in-branch usage closures were bare
/// (`"sweep [--strategy …]"`, no prefix, no program name) before cycle 0101; a
/// regression that drops the prefix again would NOT be caught by the older
/// `family_window_flag_without_real_refuses_with_usage_exit_2` pin above, which
/// only checks for the verb token and `--real` — not the `Usage: aura` lead-in.
/// An unknown `--strategy` name reaches sweep's closure; the bare verb reaches
/// walkforward's.
#[test]
fn builtin_branch_usage_lines_lead_with_the_house_style_prefix() {
    for (args, verb) in [(&["sweep", "--strategy", "bogus"][..], "sweep"), (&["walkforward"][..], "walkforward")] {
        let dir = temp_cwd("builtin_usage_prefix");
        let out = Command::new(BIN)
            .args(args)
            .current_dir(&dir)
            .output()
            .unwrap_or_else(|e| panic!("spawn aura {args:?}: {e}"));
        assert_eq!(out.status.code(), Some(2), "`aura {args:?}` must exit 2; status: {:?}", out.status);
        let stderr = String::from_utf8_lossy(&out.stderr);
        let want = format!("aura: Usage: aura {verb} ");
        assert!(stderr.starts_with(&want), "`aura {args:?}` stderr must lead with {want:?}: {stderr:?}");
        let _ = std::fs::remove_dir_all(&dir);
    }
}

/// Property (house style, #179): the built-in `mc` usage line carries TWO
/// `|`-separated grammar alternatives, and BOTH name the program (`aura mc …`),
/// not just the first. This is the one lockstep-pinned site
/// (`mc_rejects_real_flag_with_usage_exit_2` above), but that pin only checks
/// for a single `"Usage"` occurrence — a regression that fixed only the first
/// alternative (leaving the second `| aura mc <blueprint.json> --real …`
/// alternative unprefixed) would still pass it. `--seeds` with no blueprint
/// file is the built-in branch's own
/// grammar violation, reaching this exact literal. (#159 cut 4: the first
/// alternative's literal is now the blueprint-mode grammar, not a bare
/// `[--name` — `mc` is blueprint-first.)
#[test]
fn mc_builtin_usage_names_the_program_in_both_alternatives() {
    let dir = temp_cwd("mc_usage_both_alts");
    let out = Command::new(BIN)
        .args(["mc", "--seeds", "3"])
        .current_dir(&dir)
        .output()
        .expect("spawn aura mc --seeds");
    assert_eq!(out.status.code(), Some(2), "mc --seeds without a blueprint must exit 2; status: {:?}", out.status);
    let stderr = String::from_utf8_lossy(&out.stderr);
    assert!(
        stderr.contains("Usage: aura mc <blueprint.json>"),
        "first alternative must name the program: {stderr:?}"
    );
    assert!(
        stderr.contains("| aura mc <blueprint.json> --real"),
        "second alternative must ALSO name the program: {stderr:?}"
    );
    let _ = std::fs::remove_dir_all(&dir);
}

/// Property (house style, #179): `mc`'s loaded-blueprint branch usage line
/// also normalizes to `Usage: aura mc <blueprint.json> …` — before cycle 0101
/// it was fully lowercase and carried no program name (`"usage: mc
/// <blueprint.json> …"`). Reusing `sma_signal.json` (an existing closed
/// blueprint fixture) only to reach the dispatch branch; omitting `--seeds`
/// is this branch's own grammar violation.
#[test]
fn mc_over_a_blueprint_usage_names_the_program() {
    let dir = temp_cwd("mc_blueprint_usage_prefix");
    let fixture = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let out = Command::new(BIN)
        .args(["mc", &fixture])
        .current_dir(&dir)
        .output()
        .expect("spawn aura mc <blueprint.json>");
    assert_eq!(out.status.code(), Some(2), "mc <blueprint.json> without --seeds must exit 2; status: {:?}", out.status);
    let stderr = String::from_utf8_lossy(&out.stderr);
    assert!(
        stderr.contains("aura: Usage: aura mc <blueprint.json>"),
        "stderr must lead with the house-style, program-named usage: {stderr:?}"
    );
    let _ = std::fs::remove_dir_all(&dir);
}

// Note (#159 cut 4 collateral): this removes the `aura chart <name>`
// CLI-integration tests (single-run and family overlay/tap-filter/meta, the
// cross-kind write-guard collision + same-kind-overwrite pair) — every one of
// them populated its `TraceStore` fixture via a bare `run --trace`/`sweep
// --trace` invocation, both retired. The only surviving `TraceStore::write`
// call site left in the whole crate is `campaign_run::persist_campaign_traces`
// (behind a campaign's `--persist-taps`) — an entirely different, much
// heavier fixture than any of these tests set up. Removed as PIP-retirement
// collateral, not
// repointed; the render/build logic they exercised stays covered by the
// hand-built-fixture unit tests (`build_chart_data_threads_run_manifest_into_meta`
// here, `render_chart_html_*` in render.rs), which never touch the CLI or disk.

// --- `aura run --harness <name>` selector (iter-3 Task 3) --------------------

/// Property: an unknown `--harness` name is a usage error at the binary boundary —
/// exit 2 (never a silent default run), the `Err` arm wiring through to `exit(2)`.
#[test]
fn run_unknown_harness_exits_two() {
    let out = std::process::Command::new(BIN).args(["run", "--harness", "nope"]).output().unwrap();
    assert_eq!(out.status.code(), Some(2));
}

/// Property (#159 cut 2): `--strategy r-breakout` is a fully retired CLI token, on
/// BOTH `sweep` and `walkforward` — indistinguishable from an unrecognized token
/// (`bogus`). `--strategy` is now a plain `Option<String>` with no per-token
/// dispatch, so a retired value carries nothing special: it flows into clap's
/// generic usage path and out as exit 2 with the generic `aura: Usage: aura
/// <verb> …` line, never a "valid-but-unsupported" message. A regression that
/// re-taught the CLI to recognize "r-breakout" specially (rather than leaving it
/// deleted) would make this test's stderr start naming "r-breakout" instead of
/// reading the generic usage line — this is the negative that catches a
/// half-finished retirement.
#[test]
fn sweep_and_walkforward_reject_retired_r_breakout_token_as_unrecognized() {
    for (args, verb) in [
        (&["sweep", "--strategy", "r-breakout"][..], "sweep"),
        (&["walkforward", "--strategy", "r-breakout"][..], "walkforward"),
    ] {
        let cwd = temp_cwd("retired-r-breakout-token");
        let out = Command::new(BIN)
            .args(args)
            .current_dir(&cwd)
            .output()
            .unwrap_or_else(|e| panic!("spawn aura {args:?}: {e}"));
        assert_eq!(out.status.code(), Some(2), "`aura {args:?}` must exit 2; status: {:?}", out.status);
        let stderr = String::from_utf8_lossy(&out.stderr);
        if verb == "sweep" {
            let want = format!("aura: Usage: aura {verb} ");
            assert!(
                stderr.starts_with(&want),
                "`aura {args:?}` must fall into the generic usage error, not a special-cased \
                 message: {stderr:?}"
            );
        } else {
            assert!(
                stderr.contains("unexpected argument '--strategy'"),
                "`aura {args:?}` must be a clap unknown-argument rejection: {stderr:?}"
            );
        }
        assert!(out.stdout.is_empty(), "no family summary on the rejected path: {:?}", out.stdout);
        let _ = std::fs::remove_dir_all(&cwd);
    }
}

/// Property (#159 cut 3): `--strategy r-meanrev` is a fully retired token on both
/// sweep and walkforward — the generic usage error, no special-casing.
#[test]
fn sweep_and_walkforward_reject_retired_r_meanrev_token_as_unrecognized() {
    for (args, verb) in [
        (&["sweep", "--strategy", "r-meanrev"][..], "sweep"),
        (&["walkforward", "--strategy", "r-meanrev"][..], "walkforward"),
    ] {
        let cwd = temp_cwd("retired-r-meanrev-token");
        let out = Command::new(BIN).args(args).current_dir(&cwd).output()
            .unwrap_or_else(|e| panic!("spawn aura {args:?}: {e}"));
        assert_eq!(out.status.code(), Some(2), "`aura {args:?}` must exit 2; status: {:?}", out.status);
        let stderr = String::from_utf8_lossy(&out.stderr);
        if verb == "sweep" {
            let want = format!("aura: Usage: aura {verb} ");
            assert!(stderr.starts_with(&want), "generic usage, not special-cased: {stderr:?}");
        } else {
            assert!(
                stderr.contains("unexpected argument '--strategy'"),
                "`aura {args:?}` must be a clap unknown-argument rejection: {stderr:?}"
            );
        }
        assert!(out.stdout.is_empty(), "no summary on the rejected path: {:?}", out.stdout);
        let _ = std::fs::remove_dir_all(&cwd);
    }
}

/// Property (#159 cut 4): `--strategy sma` and `--strategy momentum` are fully
/// retired CLI tokens on both sweep and walkforward — the generic usage error,
/// no special-casing (the PIP built-in strategies retired with their runners).
#[test]
fn sweep_and_walkforward_reject_retired_pip_tokens_as_unrecognized() {
    for tok in ["sma", "momentum"] {
        for verb in ["sweep", "walkforward"] {
            let args = [verb, "--strategy", tok];
            let cwd = temp_cwd(&format!("retired-pip-{tok}-{verb}"));
            let out = Command::new(BIN)
                .args(args)
                .current_dir(&cwd)
                .output()
                .unwrap_or_else(|e| panic!("spawn aura {args:?}: {e}"));
            assert_eq!(
                out.status.code(),
                Some(2),
                "`aura {args:?}` must exit 2; status: {:?}",
                out.status
            );
            let stderr = String::from_utf8_lossy(&out.stderr);
            if verb == "sweep" {
                let want = format!("aura: Usage: aura {verb} ");
                assert!(
                    stderr.starts_with(&want),
                    "`aura {args:?}` must fall into the generic usage error, not a \
                     special-cased message: {stderr:?}"
                );
            } else {
                assert!(
                    stderr.contains("unexpected argument '--strategy'"),
                    "`aura {args:?}` must be a clap unknown-argument rejection: {stderr:?}"
                );
            }
            assert!(out.stdout.is_empty(), "no summary on the rejected path: {:?}", out.stdout);
            let _ = std::fs::remove_dir_all(&cwd);
        }
    }
}

/// Property (#159 cut 2): `--channel` is genuinely removed from the sweep grammar,
/// not merely unused — clap rejects it as a structurally unknown argument (exit 2)
/// rather than silently accepting and dropping it. A regression that left the flag
/// on `SweepCmd` while deleting only its plumbing would parse successfully here and
/// this test would fail on the exit code.
#[test]
fn sweep_channel_flag_is_removed_from_the_grammar() {
    let cwd = temp_cwd("sweep-channel-flag-retired");
    let out = Command::new(BIN)
        .args(["sweep", "--strategy", "sma", "--channel", "3"])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura sweep --channel");
    assert_eq!(out.status.code(), Some(2), "--channel must be an unrecognized clap argument: {:?}", out.status);
    let stderr = String::from_utf8_lossy(&out.stderr);
    assert!(
        stderr.contains("--channel") || stderr.to_lowercase().contains("unexpected argument"),
        "clap must name the unknown flag: {stderr:?}"
    );
    let _ = std::fs::remove_dir_all(&cwd);
}

/// Property (#220 generalize vertical): `--strategy`/`--fast`/`--slow` are
/// genuinely removed from the generalize grammar, not merely unused — clap
/// rejects `--strategy` as a structurally unknown argument (exit 2) rather
/// than silently accepting and ignoring it now that the candidate is an
/// arbitrary blueprint positional + generic `--axis`. A regression that left
/// the old flags on `GeneralizeCmd` while deleting only their plumbing would
/// parse successfully here and this test would fail on the exit code.
#[test]
fn generalize_strategy_flag_is_removed_from_the_grammar() {
    let cwd = temp_cwd("generalize-strategy-flag-retired");
    let out = Command::new(BIN)
        .args([
            "generalize", "--strategy", "r-sma", "--real", "GER40,USDJPY",
            "--fast", "3", "--slow", "12", "--stop-length", "14", "--stop-k", "2.0",
        ])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura generalize --strategy");
    assert_eq!(out.status.code(), Some(2), "--strategy must be an unrecognized clap argument: {:?}", out.status);
    let stderr = String::from_utf8_lossy(&out.stderr);
    assert!(
        stderr.contains("--strategy") || stderr.to_lowercase().contains("unexpected argument"),
        "clap must name the unknown flag: {stderr:?}"
    );
    let _ = std::fs::remove_dir_all(&cwd);
}

/// Property: `aura generalize` grades a single r-sma candidate across two
/// instruments — its stdout carries the per-instrument breakdown + the worst-case
/// floor + the sign-agreement, then the persisted CrossInstrument family handle.
/// Gated on local GER40/USDJPY data (the shared Sept-2024 window), skips cleanly
/// when the archive is absent.
#[test]
fn generalize_grades_a_candidate_across_two_instruments() {
    const FROM_MS: &str = "1725148800000";
    const TO_MS: &str = "1727740799999";
    let (cwd, _g) = fresh_project();
    let fixture = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args([
            "generalize", &fixture, "--real", "GER40,USDJPY",
            "--axis", "sma_signal.fast.length=3", "--axis", "sma_signal.slow.length=12",
            "--stop-length", "14", "--stop-k", "2.0",
            "--from", FROM_MS, "--to", TO_MS,
        ])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura");
    if out.status.code() == Some(1) {
        let stderr = String::from_utf8_lossy(&out.stderr);
        assert!(
            stderr.contains("no local data") || stderr.contains("no recorded geometry"),
            "exit 1 must be a data refusal, got: {stderr}"
        );
        eprintln!("skip: no local GER40/USDJPY data");
        return;
    }
    assert_eq!(out.status.code(), Some(0), "exit: {:?}", out.status);
    let stdout = String::from_utf8(out.stdout).expect("utf-8 stdout");
    assert!(stdout.contains("\"generalize\":"), "aggregate object: {stdout}");
    assert!(stdout.contains("\"n_instruments\":2"), "two instruments: {stdout}");
    assert!(stdout.contains("\"worst_case\":"), "worst_case present: {stdout}");
    assert!(stdout.contains("\"sign_agreement\":"), "sign_agreement present: {stdout}");
    assert!(stdout.contains("GER40") && stdout.contains("USDJPY"), "per-instrument breakdown: {stdout}");
    assert!(stdout.contains("\"family_id\":\"generalize-0\""), "family handle: {stdout}");
}

/// Characterization pin (byte-identity anchor for the generalize dissolution,
/// #210). The current `aura generalize` binds its protective stop as a grid axis
/// (`blueprint_sweep_family` with `stop_open=true`); the dissolution rebinds the same
/// stop through the risk-regime seam (`RiskRegime::Vol -> StopRule::Vol`). The
/// sibling `generalize_grades_a_candidate_across_two_instruments` asserts shape
/// only — `worst_case` and the per-instrument R floats are never checked — so a
/// stop-mechanism divergence between the two bindings would ship green. This pins
/// the EXACT current R grade of the identical invocation; after the dissolution
/// the same command must reproduce these bytes (the acceptance gate), and any
/// drift fails here loudly. Gated on the shared GER40/USDJPY Sept-2024 archive;
/// skips cleanly on a data refusal.
#[test]
fn generalize_real_e2e_pins_the_exact_current_grade() {
    const FROM_MS: &str = "1725148800000";
    const TO_MS: &str = "1727740799999";
    let (cwd, _g) = fresh_project();
    let fixture = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args([
            "generalize", &fixture, "--real", "GER40,USDJPY",
            "--axis", "sma_signal.fast.length=3", "--axis", "sma_signal.slow.length=12",
            "--stop-length", "14", "--stop-k", "2.0",
            "--from", FROM_MS, "--to", TO_MS,
        ])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura");
    if out.status.code() == Some(1) {
        let stderr = String::from_utf8_lossy(&out.stderr);
        assert!(
            stderr.contains("no local data") || stderr.contains("no recorded geometry"),
            "exit 1 must be a data refusal, got: {stderr}"
        );
        eprintln!("skip: no local GER40/USDJPY data");
        return;
    }
    assert_eq!(out.status.code(), Some(0), "exit: {:?}", out.status);
    let stdout = String::from_utf8(out.stdout).expect("utf-8 stdout");
    let grade_line = stdout
        .lines()
        .find(|l| l.starts_with("{\"generalize\":"))
        .unwrap_or_else(|| panic!("the generalize grade line: {stdout}"));
    let v: serde_json::Value = serde_json::from_str(grade_line).expect("grade line parses as JSON");
    let g = &v["generalize"];
    assert_eq!(g["metric"].as_str(), Some("expectancy_r"), "metric: {grade_line}");
    assert_eq!(g["n_instruments"].as_u64(), Some(2), "two instruments: {grade_line}");
    assert_eq!(g["sign_agreement"].as_u64(), Some(2), "both agree in sign: {grade_line}");
    // EXACT R floats — the byte-identity anchor the dissolution must preserve.
    assert_eq!(g["worst_case"].as_f64(), Some(0.005795903617609842), "worst-case R floor: {grade_line}");
    let per = g["per_instrument"].as_array().expect("per_instrument is an array");
    assert_eq!(per.len(), 2, "two per-instrument grades: {grade_line}");
    assert_eq!(per[0][0].as_str(), Some("GER40"), "first instrument: {grade_line}");
    assert_eq!(per[0][1].as_f64(), Some(0.01056371324510624), "GER40 expectancy R: {grade_line}");
    assert_eq!(per[1][0].as_str(), Some("USDJPY"), "second instrument: {grade_line}");
    assert_eq!(per[1][1].as_f64(), Some(0.005795903617609842), "USDJPY expectancy R: {grade_line}");
}

/// Property (#220 generalize vertical): `aura generalize` grades ANY sweepable
/// blueprint, not just the historical r-sma candidate — the whole point of
/// dropping `--strategy r-sma`/`--fast`/`--slow` for a positional blueprint +
/// generic `--axis`. Every other generalize E2E test here still exercises
/// `r_sma.json`, so a regression that silently special-cased r-sma paths
/// back in (e.g. resolving axes against a hardcoded `sma_signal.*` namespace)
/// would ship green there while breaking every other blueprint. This runs the
/// unrelated r-breakout candidate (the ganged `channel_length` axis, no
/// `sma_signal` node in sight) end-to-end and asserts a well-formed two-
/// instrument grade. Gated on the shared GER40/USDJPY Sept-2024 archive; skips
/// cleanly on a data refusal.
#[test]
fn generalize_grades_a_non_r_sma_blueprint() {
    const FROM_MS: &str = "1725148800000";
    const TO_MS: &str = "1727740799999";
    let (cwd, _g) = fresh_project();
    let fixture = format!("{}/tests/fixtures/r_breakout_open.json", env!("CARGO_MANIFEST_DIR"));
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args([
            "generalize", &fixture, "--real", "GER40,USDJPY",
            "--axis", "r_breakout_signal.channel_length=20",
            "--stop-length", "14", "--stop-k", "2.0",
            "--from", FROM_MS, "--to", TO_MS,
        ])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura");
    if out.status.code() == Some(1) {
        let stderr = String::from_utf8_lossy(&out.stderr);
        assert!(
            stderr.contains("no local data") || stderr.contains("no recorded geometry"),
            "exit 1 must be a data refusal, got: {stderr}"
        );
        eprintln!("skip: no local GER40/USDJPY data");
        return;
    }
    assert_eq!(out.status.code(), Some(0), "exit: {:?}", out.status);
    let stdout = String::from_utf8(out.stdout).expect("utf-8 stdout");
    assert!(stdout.contains("\"generalize\":"), "aggregate object: {stdout}");
    assert!(stdout.contains("\"n_instruments\":2"), "two instruments: {stdout}");
    assert!(stdout.contains("\"worst_case\":"), "worst_case present: {stdout}");
    assert!(stdout.contains("\"sign_agreement\":"), "sign_agreement present: {stdout}");
    assert!(stdout.contains("GER40") && stdout.contains("USDJPY"), "per-instrument breakdown: {stdout}");
    assert!(stdout.contains("\"family_id\":"), "family handle: {stdout}");
}

/// Characterization pin (byte-identity anchor for the walkforward dissolution,
/// #210). The current `aura walkforward` runs its per-window IS-refit inline
/// (`walkforward_family` -> `select_winner` -> OOS run -> stitch/pool); the
/// dissolution reroutes the same rolling IS-sweep -> OOS-run -> deflation through
/// the campaign `std::walk_forward` stage, and must retain the OOS pip curve the
/// campaign runner currently discards (`oos_equity: vec![]`) to reproduce
/// `stitched_total_pips`, plus reconstruct pooled `oos_r`. The multi-point grid
/// (fast 3,5 x slow 12,20) makes the per-window winner selection non-degenerate --
/// `param_stability` varies across windows -- so a winner-pick divergence between
/// the inline and campaign refit cannot ship green. This pins the EXACT current
/// grade of the identical invocation over a fixed 2025 GER40 window; after the
/// dissolution the same command must reproduce these bytes (the acceptance gate).
/// Gated on the shared GER40 archive; skips cleanly on a data refusal.
#[test]
fn walkforward_real_e2e_pins_the_exact_current_grade() {
    const FROM_MS: &str = "1735689600000";
    const TO_MS: &str = "1767225599000";
    let (cwd, _g) = fresh_project();
    let fixture = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args([
            "walkforward", &fixture, "--real", "GER40",
            "--axis", "sma_signal.fast.length=3,5", "--axis", "sma_signal.slow.length=12,20",
            "--stop-length", "14", "--stop-k", "2.0",
            "--from", FROM_MS, "--to", TO_MS,
        ])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura");
    if out.status.code() == Some(1) {
        let stderr = String::from_utf8_lossy(&out.stderr);
        assert!(
            stderr.contains("no local data") || stderr.contains("no recorded geometry"),
            "exit 1 must be a data refusal, got: {stderr}"
        );
        eprintln!("skip: no local GER40 data");
        return;
    }
    assert_eq!(out.status.code(), Some(0), "exit: {:?}", out.status);
    let stdout = String::from_utf8(out.stdout).expect("utf-8 stdout");
    let grade_line = stdout
        .lines()
        .find(|l| l.starts_with("{\"walkforward\":"))
        .unwrap_or_else(|| panic!("the walkforward summary line: {stdout}"));
    let v: serde_json::Value = serde_json::from_str(grade_line).expect("summary line parses as JSON");
    let wf = &v["walkforward"];
    assert_eq!(wf["windows"].as_u64(), Some(9), "window count: {grade_line}");
    // EXACT floats -- the byte-identity anchor the dissolution must preserve.
    // stitched_total_pips is retained only if the campaign runner keeps the OOS curve.
    assert_eq!(wf["stitched_total_pips"].as_f64(), Some(-1039.8606666650755), "stitched OOS pips: {grade_line}");
    let r = &wf["oos_r"];
    assert_eq!(r["n_trades"].as_u64(), Some(20681), "pooled OOS trade count: {grade_line}");
    assert_eq!(r["expectancy_r"].as_f64(), Some(-0.002397100685333715), "pooled OOS expectancy R: {grade_line}");
    // param_stability varies across windows -> the per-window IS-refit picks
    // different winners; pinning the means locks the winner-selection equivalence.
    let ps = wf["param_stability"].as_array().expect("param_stability is an array");
    assert_eq!(ps[0]["mean"].as_f64(), Some(3.888888888888889), "fast-MA refit mean: {grade_line}");
    assert_eq!(ps[1]["mean"].as_f64(), Some(17.333333333333332), "slow-MA refit mean: {grade_line}");
}

/// Property (#220 walkforward vertical): `aura walkforward --real` IS-refits ANY
/// sweepable blueprint through the campaign path, not just the historical r-sma
/// candidate — the whole point of dropping `--strategy r-sma`/`--fast`/`--slow`
/// for a positional blueprint + generic `--axis`. Every other walkforward E2E
/// test here still exercises `r_sma.json`; a regression that silently
/// special-cased the refit's axis lookup back to a hardcoded `sma_signal.*`
/// namespace (e.g. in `walkforward_summary_json_from_reports`'s per-axis
/// `raw_matches_wrapped` search) would ship green there while breaking every
/// other blueprint. This runs the unrelated r-breakout candidate
/// (the ganged `channel_length` axis, no `sma_signal` node in sight) end-to-end
/// and asserts a well-formed multi-window summary whose `param_stability` has
/// exactly one entry per bound axis (1 ganged blueprint axis + the 2 stop columns).
/// The window count (9) is roller/data-derived, not strategy-derived, so it is
/// pinned exactly against the same GER40 2025 span the r-sma anchor uses.
/// Gated on the shared GER40 archive; skips cleanly on a data refusal.
#[test]
fn walkforward_dissolves_a_non_r_sma_blueprint() {
    const FROM_MS: &str = "1735689600000";
    const TO_MS: &str = "1767225599000";
    let (cwd, _g) = fresh_project();
    let fixture = format!("{}/tests/fixtures/r_breakout_open.json", env!("CARGO_MANIFEST_DIR"));
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args([
            "walkforward", &fixture, "--real", "GER40",
            "--axis", "r_breakout_signal.channel_length=10,20",
            "--stop-length", "14", "--stop-k", "2.0",
            "--from", FROM_MS, "--to", TO_MS,
        ])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura");
    if out.status.code() == Some(1) {
        let stderr = String::from_utf8_lossy(&out.stderr);
        assert!(
            stderr.contains("no local data") || stderr.contains("no recorded geometry"),
            "exit 1 must be a data refusal, got: {stderr}"
        );
        eprintln!("skip: no local GER40 data");
        return;
    }
    assert_eq!(out.status.code(), Some(0), "exit: {:?}", out.status);
    let stdout = String::from_utf8(out.stdout).expect("utf-8 stdout");
    let grade_line = stdout
        .lines()
        .find(|l| l.starts_with("{\"walkforward\":"))
        .unwrap_or_else(|| panic!("the walkforward summary line: {stdout}"));
    let v: serde_json::Value = serde_json::from_str(grade_line).expect("summary line parses as JSON");
    let wf = &v["walkforward"];
    assert_eq!(wf["windows"].as_u64(), Some(9), "window count (data/roller-derived): {grade_line}");
    assert!(wf["oos_r"]["n_trades"].as_u64().is_some(), "pooled OOS trade count present: {grade_line}");
    let ps = wf["param_stability"].as_array().expect("param_stability is an array");
    assert_eq!(ps.len(), 3, "one entry per bound axis (channel_length, stop_length, stop_k): {grade_line}");
}

/// Fork A: the dissolved walkforward binds the stop as a single risk regime, so a
/// multi-value --stop-length/--stop-k is refused (exit 2) before any run.
#[test]
fn walkforward_dissolved_refuses_a_multi_value_stop() {
    let cwd = temp_cwd("walkforward-multi-stop");
    let fixture = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args([
            "walkforward", &fixture, "--real", "GER40",
            "--axis", "sma_signal.fast.length=3", "--axis", "sma_signal.slow.length=12",
            "--stop-length", "14,20", "--stop-k", "2.0",
        ])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura");
    assert_eq!(out.status.code(), Some(2), "multi-value stop is a usage refusal");
    let stderr = String::from_utf8_lossy(&out.stderr);
    assert!(stderr.contains("single risk regime"), "stop-regime refusal: {stderr}");
}

/// #215: --select plateau is no longer refused on the dissolved path — it now
/// threads through to the wf stage exactly like the built-in synthetic path.
/// The property is flag acceptance, not window semantics, so the window is
/// bounded (~135 days, one IS(90d)+OOS(30d) roll) rather than streaming the
/// full multi-year archive. Gated on the shared GER40 archive; skips cleanly
/// on a data refusal (exit 1), same tolerance as the sibling
/// stop-knob-defaulting test.
#[test]
fn walkforward_dissolved_accepts_select_plateau() {
    let (cwd, _g) = fresh_project();
    let fixture = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args([
            "walkforward", &fixture, "--real", "GER40",
            "--axis", "sma_signal.fast.length=3", "--axis", "sma_signal.slow.length=12",
            "--stop-length", "14", "--stop-k", "2.0",
            "--select", "plateau:worst",
            "--from", "1735689600000", "--to", "1747353600000",
        ])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura");
    if out.status.code() == Some(1) {
        let stderr = String::from_utf8_lossy(&out.stderr);
        assert!(
            stderr.contains("no local data") || stderr.contains("no recorded geometry"),
            "exit 1 must be a data refusal, got: {stderr}"
        );
        eprintln!("skip: no local GER40 data");
        return;
    }
    assert_eq!(
        out.status.code(),
        Some(0),
        "--select plateau:worst must no longer be refused on the dissolved path: {}",
        String::from_utf8_lossy(&out.stderr)
    );
}

/// Characterization pin (#215): the same identical invocation as
/// `walkforward_real_e2e_pins_the_exact_current_grade` (fixture, real
/// instrument, axes, stop, window), but selecting `plateau:worst` instead of
/// the default argmax. Plateau scores the closed neighbourhood over the full
/// IS grid rather than picking the single best point, so its winner (and
/// therefore its OOS stitch) legitimately differs from the argmax anchor;
/// this pins that divergence exactly, so a regression that silently
/// collapsed plateau back to argmax (or vice versa) inside
/// `run_walk_forward_stage`'s `select_sweep_winner` dispatch goes red here
/// even though the argmax anchor stays green. Gated on the shared GER40
/// archive; skips cleanly on a data refusal.
#[test]
fn walkforward_real_e2e_pins_the_exact_current_plateau_grade() {
    const FROM_MS: &str = "1735689600000";
    const TO_MS: &str = "1767225599000";
    let (cwd, _g) = fresh_project();
    let fixture = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args([
            "walkforward", &fixture, "--real", "GER40",
            "--axis", "sma_signal.fast.length=3,5", "--axis", "sma_signal.slow.length=12,20",
            "--stop-length", "14", "--stop-k", "2.0",
            "--from", FROM_MS, "--to", TO_MS,
            "--select", "plateau:worst",
        ])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura");
    if out.status.code() == Some(1) {
        let stderr = String::from_utf8_lossy(&out.stderr);
        assert!(
            stderr.contains("no local data") || stderr.contains("no recorded geometry"),
            "exit 1 must be a data refusal, got: {stderr}"
        );
        eprintln!("skip: no local GER40 data");
        return;
    }
    assert_eq!(out.status.code(), Some(0), "exit: {:?}", out.status);
    let stdout = String::from_utf8(out.stdout).expect("utf-8 stdout");
    let grade_line = stdout
        .lines()
        .find(|l| l.starts_with("{\"walkforward\":"))
        .unwrap_or_else(|| panic!("the walkforward summary line: {stdout}"));
    let v: serde_json::Value = serde_json::from_str(grade_line).expect("summary line parses as JSON");
    let wf = &v["walkforward"];
    assert_eq!(wf["windows"].as_u64(), Some(9), "window count: {grade_line}");
    // EXACT floats -- the byte-identity anchor a plateau-vs-argmax regression must preserve.
    assert_eq!(wf["stitched_total_pips"].as_f64(), Some(-968.3776666637464), "stitched OOS pips: {grade_line}");
    let r = &wf["oos_r"];
    assert_eq!(r["n_trades"].as_u64(), Some(20667), "pooled OOS trade count: {grade_line}");
    assert_eq!(r["expectancy_r"].as_f64(), Some(0.007635120926372627), "pooled OOS expectancy R: {grade_line}");
    // param_stability varies across windows -> the per-window plateau refit picks
    // different winners than the argmax anchor; pinning the means locks the
    // plateau winner-selection path.
    let ps = wf["param_stability"].as_array().expect("param_stability is an array");
    assert_eq!(ps[0]["mean"].as_f64(), Some(4.777777777777778), "fast-MA refit mean: {grade_line}");
    assert_eq!(ps[1]["mean"].as_f64(), Some(16.444444444444443), "slow-MA refit mean: {grade_line}");
}

/// Property (#217): `walkforward_args_from` no longer requires both stop knobs
/// together on the campaign path — each defaults independently to the
/// single-sourced regime. Omitting only `--stop-k` while keeping `--stop-length`
/// explicit still succeeds, proving the defaulting is per-flag, not
/// all-or-nothing (the sibling
/// `walkforward_stopless_defaults_the_regime_byte_identically_to_explicit_3_2`
/// pins the both-omitted case). The property is per-flag defaulting, not
/// window semantics, so the window is bounded (~135 days) rather than
/// streaming the full multi-year archive. Gated on the shared GER40 archive;
/// skips cleanly on a data refusal.
#[test]
fn walkforward_dissolved_defaults_a_single_omitted_knob() {
    let (cwd, _g) = fresh_project();
    let fixture = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args([
            "walkforward", &fixture, "--real", "GER40",
            "--axis", "sma_signal.fast.length=3", "--axis", "sma_signal.slow.length=12",
            "--stop-length", "3",
            "--from", "1735689600000", "--to", "1747353600000",
        ])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura");
    if out.status.code() == Some(1) {
        let stderr = String::from_utf8_lossy(&out.stderr);
        assert!(
            stderr.contains("no local data") || stderr.contains("no recorded geometry"),
            "exit 1 must be a data refusal, got: {stderr}"
        );
        eprintln!("skip: no local GER40 data");
        return;
    }
    assert_eq!(
        out.status.code(),
        Some(0),
        "omitting only --stop-k must still succeed (it defaults independently): {}",
        String::from_utf8_lossy(&out.stderr)
    );
}

/// An empty `--real ""` still satisfies clap's `Option::is_some()` dispatch guard, so
/// `walkforward_args_from`'s own empty-string check is what refuses it (exit 2) —
/// distinct from omitting `--real` entirely, which never enters the dissolved branch.
#[test]
fn walkforward_dissolved_refuses_an_empty_real_symbol() {
    let cwd = temp_cwd("walkforward-empty-real");
    let fixture = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args([
            "walkforward", &fixture, "--real", "",
            "--axis", "sma_signal.fast.length=3", "--axis", "sma_signal.slow.length=12",
            "--stop-length", "14", "--stop-k", "2.0",
        ])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura");
    assert_eq!(out.status.code(), Some(2), "empty --real is a usage refusal");
    let stderr = String::from_utf8_lossy(&out.stderr);
    assert!(stderr.contains("dissolves only over --real"), "empty-real refusal: {stderr}");
}

/// Property: an `--axis` name that is not among the loaded blueprint's sweepable
/// axes is refused before the raw-namespace strip or any archive access (exit 2),
/// echoing exactly the name the user typed — the same WRAPPED-probe preflight the
/// sibling verbs enforce
/// (`aura_sweep_real_blueprint_refuses_a_raw_form_axis_name_before_data_access`,
/// `generalize_refuses_an_unknown_axis_name`). Data-free: no archive/store access.
#[test]
fn walkforward_dissolved_refuses_an_unknown_axis_name() {
    let cwd = temp_cwd("walkforward-unknown-axis");
    let fixture = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args([
            "walkforward", &fixture, "--real", "GER40",
            "--axis", "sma_signal.nope=3", "--axis", "sma_signal.slow.length=12",
            "--stop-length", "14", "--stop-k", "2.0",
        ])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura");
    assert_eq!(out.status.code(), Some(2), "an unknown axis name must exit 2");
    let stderr = String::from_utf8_lossy(&out.stderr);
    assert!(
        stderr.contains("axis \"sma_signal.nope\"") && stderr.contains("sweepable axes"),
        "must name the unresolved axis exactly as typed: {stderr}"
    );
    assert!(
        out.stdout.is_empty(),
        "the refusal must not leak an aggregate to stdout, got: {:?}",
        String::from_utf8_lossy(&out.stdout)
    );
    assert!(!cwd.join("runs").exists(), "a refused axis name must not start a real run");
}

/// Front-end parity: an unknown `--select` token is refused (exit 2) on the
/// dissolved path exactly as it is on the inline path — it must not be silently
/// swallowed by `select_rule_of`'s conversion and fall through to argmax.
#[test]
fn walkforward_dissolved_refuses_an_unknown_select_token() {
    let cwd = temp_cwd("walkforward-bad-select");
    let fixture = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args([
            "walkforward", &fixture, "--real", "GER40",
            "--axis", "sma_signal.fast.length=3", "--axis", "sma_signal.slow.length=12",
            "--stop-length", "14", "--stop-k", "2.0",
            "--select", "bogus",
        ])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura");
    assert_eq!(out.status.code(), Some(2), "unknown --select token is a usage refusal");
    let stderr = String::from_utf8_lossy(&out.stderr);
    assert!(stderr.contains("Usage: aura walkforward"), "unknown-select refusal: {stderr}");
}

/// `--name`/`--trace` are mutually exclusive on the walkforward real-data path
/// (#224 delivers `--trace` there — the up-front #168 forward-pointer refusal
/// is gone; `walkforward_args_from`'s own mutual-exclusion check is what fires
/// when both are given at once), exit 2, naming the mutual-exclusion contract
/// rather than a now-stale #224 forward pointer.
#[test]
fn walkforward_dissolved_refuses_name_and_trace_together() {
    let cwd = temp_cwd("walkforward-name-and-trace");
    let fixture = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args([
            "walkforward", &fixture, "--real", "GER40",
            "--axis", "sma_signal.fast.length=3", "--axis", "sma_signal.slow.length=12",
            "--stop-length", "14", "--stop-k", "2.0",
            "--name", "a", "--trace", "b",
        ])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura");
    assert_eq!(out.status.code(), Some(2), "--name/--trace together is refused (exit 2)");
    let stderr = String::from_utf8_lossy(&out.stderr);
    assert!(
        stderr.contains("mutually exclusive"),
        "refusal names the mutual-exclusion contract: {stderr}"
    );
}

/// Property (#210 T3, dispatch split): a successful real-data walkforward
/// (`aura walkforward <blueprint> --real --axis …`; spelled `--strategy r-sma
/// --real` before the #220 argv migration) durably auto-registers exactly one
/// generated process document, one
/// generated campaign document (carrying the `--name` handle and the stop as a
/// non-empty single risk regime), and one campaign-run record — mirroring the
/// generalize dissolution's audit trail. The persisted family set is exactly one
/// `WalkForward` family (the per-window OOS reports) and NO `Sweep` family —
/// the leading `std::grid` stage only enumerates (#256) — the observable proof that
/// `dispatch_walkforward`'s real-data path runs through the one campaign
/// executor rather than the deleted inline roller. Gated on the shared GER40
/// archive; skips cleanly on a data refusal.
#[test]
fn walkforward_dissolves_through_the_campaign_path() {
    const FROM_MS: &str = "1735689600000";
    const TO_MS: &str = "1767225599000";
    let (cwd, _g) = fresh_project();
    let fixture = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args([
            "walkforward", &fixture, "--real", "GER40",
            "--axis", "sma_signal.fast.length=3", "--axis", "sma_signal.slow.length=12",
            "--stop-length", "14", "--stop-k", "2.0",
            "--from", FROM_MS, "--to", TO_MS,
        ])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura");
    if out.status.code() == Some(1) {
        let stderr = String::from_utf8_lossy(&out.stderr);
        assert!(
            stderr.contains("no local data") || stderr.contains("no recorded geometry"),
            "exit 1 must be a data refusal, got: {stderr}"
        );
        eprintln!("skip: no local GER40 data");
        return;
    }
    assert_eq!(out.status.code(), Some(0), "exit: {:?}", out.status);

    let count = |sub: &str| {
        std::fs::read_dir(cwd.join("runs").join(sub))
            .map(|d| d.count())
            .unwrap_or(0)
    };
    assert_eq!(count("processes"), 1, "one generated process document registered");
    assert_eq!(count("campaigns"), 1, "one generated campaign document registered");
    let runs_log = std::fs::read_to_string(cwd.join("runs").join("campaign_runs.jsonl"))
        .expect("campaign_runs.jsonl exists after a sugar run");
    assert_eq!(runs_log.lines().count(), 1, "one campaign run recorded");

    let campaigns_dir = cwd.join("runs").join("campaigns");
    let campaign_doc_path = std::fs::read_dir(&campaigns_dir)
        .expect("campaigns dir exists")
        .next()
        .expect("exactly one campaign document")
        .expect("readable dir entry")
        .path();
    let campaign_doc_json = std::fs::read_to_string(&campaign_doc_path).expect("read campaign doc");
    assert!(
        campaign_doc_json.contains("\"name\":\"walkforward\""),
        "the generated campaign document carries the --name handle: {campaign_doc_json}"
    );
    assert!(
        campaign_doc_json.contains("\"risk\":[")
            && campaign_doc_json.contains("\"length\":14")
            && campaign_doc_json.contains("\"k\":2.0"),
        "the stop rides a non-empty risk regime: {campaign_doc_json}"
    );

    let fams = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args(["runs", "families"])
        .current_dir(&cwd)
        .output()
        .expect("spawn families");
    let fams_out = String::from_utf8_lossy(&fams.stdout).into_owned();
    assert_eq!(
        fams_out.lines().filter(|l| l.contains("\"kind\":\"Sweep\"")).count(),
        0,
        "the enumerate-only std::grid leading stage persists no Sweep family: {fams_out}"
    );
    assert_eq!(
        fams_out.lines().filter(|l| l.contains("\"kind\":\"WalkForward\"")).count(),
        1,
        "one WalkForward family holding the per-window OOS reports: {fams_out}"
    );
}

/// Property (#217): on the dissolved `aura walkforward --real` path,
/// `--stop-length`/`--stop-k` are OPTIONAL — a stop-less invocation succeeds,
/// defaulting each omitted knob to the single-sourced default regime (length 3,
/// k 2.0), the very regime the campaign member runner already resolves for an
/// unbound regime (`stop_rule_for_regime`). The default is byte-identical to
/// spelling the knobs out: running the same blueprint in the same store first
/// stop-less, then with an explicit `--stop-length 3 --stop-k 2.0`, leaves
/// EXACTLY ONE generated campaign document — the two documents are byte-identical
/// so content-addressed storage collapses them onto one content id — while each
/// invocation still records its own campaign-run line. This aligns walkforward's
/// knob contract with sweep's (which already defaults the stop) without any
/// content-id churn on the explicit spelling. The sibling verbs mc/generalize gain
/// the same independent per-flag defaulting on the GREEN side. Gated on the shared
/// GER40 archive; skips cleanly on a data refusal. (The multi-value stop is still
/// refused — `walkforward_dissolved_refuses_a_multi_value_stop` — the stop stays a
/// single risk regime; only its *presence* becomes optional.)
#[test]
fn walkforward_stopless_defaults_the_regime_byte_identically_to_explicit_3_2() {
    const FROM_MS: &str = "1735689600000";
    const TO_MS: &str = "1767225599000";
    let (cwd, _g) = fresh_project();
    let fixture = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));

    // Stop-less: the two knobs are omitted entirely. This is the headline — today it
    // is refused at the argv boundary (exit 2, "requires --stop-length --stop-k");
    // the feature makes the knobs optional so this run defaults the regime and runs.
    let stopless = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args([
            "walkforward", &fixture, "--real", "GER40",
            "--axis", "sma_signal.fast.length=3", "--axis", "sma_signal.slow.length=12",
            "--from", FROM_MS, "--to", TO_MS,
        ])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura");
    if stopless.status.code() == Some(1) {
        let stderr = String::from_utf8_lossy(&stopless.stderr);
        assert!(
            stderr.contains("no local data") || stderr.contains("no recorded geometry"),
            "exit 1 must be a data refusal, got: {stderr}"
        );
        eprintln!("skip: no local GER40 data");
        return;
    }
    assert_eq!(
        stopless.status.code(),
        Some(0),
        "a stop-less walkforward must succeed (the stop knobs are optional, each defaulting \
         to the single-sourced 3/2.0 regime): {}",
        String::from_utf8_lossy(&stopless.stderr)
    );

    // Explicit 3/2.0: names the very defaults the stop-less run resolves. Same
    // blueprint, same axes, same default `--name` ("walkforward"), same window.
    let explicit = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args([
            "walkforward", &fixture, "--real", "GER40",
            "--axis", "sma_signal.fast.length=3", "--axis", "sma_signal.slow.length=12",
            "--stop-length", "3", "--stop-k", "2.0",
            "--from", FROM_MS, "--to", TO_MS,
        ])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura");
    assert_eq!(
        explicit.status.code(),
        Some(0),
        "the explicit --stop-length 3 --stop-k 2.0 walkforward must succeed: {}",
        String::from_utf8_lossy(&explicit.stderr)
    );

    // Byte-identity: the default regime IS 3/2.0, so both invocations generate the
    // same campaign document and content-addressed storage collapses them onto one
    // content id — exactly one campaign document in the store.
    let count = |sub: &str| {
        std::fs::read_dir(cwd.join("runs").join(sub))
            .map(|d| d.count())
            .unwrap_or(0)
    };
    assert_eq!(
        count("campaigns"),
        1,
        "the stop-less default binds the SAME 3/2.0 regime as the explicit run, so the two \
         campaign documents are byte-identical and dedup to one content id"
    );
    let runs_log = std::fs::read_to_string(cwd.join("runs").join("campaign_runs.jsonl"))
        .expect("campaign_runs.jsonl exists after two sugar runs");
    assert_eq!(
        runs_log.lines().count(),
        2,
        "each invocation still records its own campaign run (a run is an event, a document \
         is content)"
    );
}

/// Property: `aura generalize` is now thin sugar over the one campaign path — a
/// successful run durably auto-registers exactly one generated process document,
/// one generated campaign document (carrying the `--name` handle and the stop as
/// a non-empty single risk regime), and one campaign-run record. This is the
/// observable proof the inline path is gone and the dissolution runs through the
/// executor. Gated on the shared GER40/USDJPY Sept-2024 archive; skips cleanly on
/// a data refusal.
#[test]
fn generalize_dissolves_through_the_campaign_path() {
    const FROM_MS: &str = "1725148800000";
    const TO_MS: &str = "1727740799999";
    let (cwd, _g) = fresh_project();
    let fixture = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args([
            "generalize", &fixture, "--real", "GER40,USDJPY",
            "--axis", "sma_signal.fast.length=3", "--axis", "sma_signal.slow.length=12",
            "--stop-length", "14", "--stop-k", "2.0",
            "--from", FROM_MS, "--to", TO_MS,
        ])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura");
    if out.status.code() == Some(1) {
        let stderr = String::from_utf8_lossy(&out.stderr);
        assert!(
            stderr.contains("no local data") || stderr.contains("no recorded geometry"),
            "exit 1 must be a data refusal, got: {stderr}"
        );
        eprintln!("skip: no local GER40/USDJPY data");
        return;
    }
    assert_eq!(out.status.code(), Some(0), "exit: {:?}", out.status);

    let count = |sub: &str| {
        std::fs::read_dir(cwd.join("runs").join(sub))
            .map(|d| d.count())
            .unwrap_or(0)
    };
    assert_eq!(count("processes"), 1, "one generated process document registered");
    assert_eq!(count("campaigns"), 1, "one generated campaign document registered");
    let runs_log = std::fs::read_to_string(cwd.join("runs").join("campaign_runs.jsonl"))
        .expect("campaign_runs.jsonl exists after a sugar run");
    assert_eq!(runs_log.lines().count(), 1, "one campaign run recorded");

    let campaigns_dir = cwd.join("runs").join("campaigns");
    let campaign_doc_path = std::fs::read_dir(&campaigns_dir)
        .expect("campaigns dir exists")
        .next()
        .expect("exactly one campaign document")
        .expect("readable dir entry")
        .path();
    let campaign_doc_json = std::fs::read_to_string(&campaign_doc_path).expect("read campaign doc");
    assert!(
        campaign_doc_json.contains("\"name\":\"generalize\""),
        "the generated campaign document carries the --name handle: {campaign_doc_json}"
    );
    // The structural delta vs the sweep translator: the stop is a non-empty
    // single risk regime, not empty risk.
    assert!(
        campaign_doc_json.contains("\"risk\":[")
            && campaign_doc_json.contains("\"length\":14")
            && campaign_doc_json.contains("\"k\":2.0"),
        "the stop rides a non-empty risk regime: {campaign_doc_json}"
    );

    // The campaign path's family set (#210 Q4): exactly one CrossInstrument grade
    // family (the generalize result, appended by the sugar) plus one
    // per-instrument Sweep family per cell (persisted by the executor).
    let fams = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args(["runs", "families"])
        .current_dir(&cwd)
        .output()
        .expect("spawn families");
    let fams_out = String::from_utf8_lossy(&fams.stdout).into_owned();
    assert_eq!(
        fams_out.lines().filter(|l| l.contains("\"kind\":\"CrossInstrument\"")).count(),
        1,
        "one CrossInstrument grade family: {fams_out}"
    );
    assert_eq!(
        fams_out.lines().filter(|l| l.contains("\"kind\":\"Sweep\"")).count(),
        2,
        "one per-instrument Sweep family per cell: {fams_out}"
    );
}

/// Property (#210, dissolution — content-addressed idempotency): re-running the
/// IDENTICAL `aura generalize` invocation in the same project does not litter the
/// store with a second generated process/campaign document — `register_generated_g`
/// writes under a content-id filename, so two runs of the same candidate collapse
/// onto the same one process doc and one campaign doc, while each invocation still
/// records its OWN campaign-run line (a run is an event, a document is content).
/// Were the sugar to salt its generated documents with anything non-deterministic
/// (e.g. a timestamp), this would regress from 1 to 2 stored documents per repeat
/// invocation — the store-litter regression the campaign-path dissolution must not
/// introduce. Gated on the shared GER40/USDJPY Sept-2024 archive; skips cleanly on
/// a data refusal.
#[test]
fn generalize_repeated_identical_invocation_does_not_litter_the_store() {
    const FROM_MS: &str = "1725148800000";
    const TO_MS: &str = "1727740799999";
    let (cwd, _g) = fresh_project();
    let fixture = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let invoke = || {
        std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
            .args([
                "generalize", &fixture, "--real", "GER40,USDJPY",
                "--axis", "sma_signal.fast.length=3", "--axis", "sma_signal.slow.length=12",
                "--stop-length", "14", "--stop-k", "2.0",
                "--from", FROM_MS, "--to", TO_MS,
            ])
            .current_dir(&cwd)
            .output()
            .expect("spawn aura")
    };
    let first = invoke();
    if first.status.code() == Some(1) {
        let stderr = String::from_utf8_lossy(&first.stderr);
        assert!(
            stderr.contains("no local data") || stderr.contains("no recorded geometry"),
            "exit 1 must be a data refusal, got: {stderr}"
        );
        eprintln!("skip: no local GER40/USDJPY data");
        return;
    }
    assert_eq!(first.status.code(), Some(0), "first run exit: {:?}", first.status);
    let second = invoke();
    assert_eq!(second.status.code(), Some(0), "second run exit: {:?}", second.status);

    let count = |sub: &str| {
        std::fs::read_dir(cwd.join("runs").join(sub))
            .map(|d| d.count())
            .unwrap_or(0)
    };
    assert_eq!(count("processes"), 1, "two identical invocations share one process document");
    assert_eq!(count("campaigns"), 1, "two identical invocations share one campaign document");
    let runs_log = std::fs::read_to_string(cwd.join("runs").join("campaign_runs.jsonl"))
        .expect("campaign_runs.jsonl exists");
    assert_eq!(runs_log.lines().count(), 2, "each invocation still records its own campaign run");
}

/// Property (#210, dissolution — content-addressed distinctness): the idempotency
/// above is not a bug masquerading as a feature — two DIFFERENT `--fast`
/// invocations (differing candidate content) generate and persist TWO distinct
/// campaign documents (and two distinct process documents, since the `--metric`
/// text is unchanged so the process content matches — pinned via the campaign
/// count only, the axis differs on the campaign, not the process). Together with
/// `generalize_repeated_identical_invocation_does_not_litter_the_store`, this rules
/// out a store that always overwrites one fixed filename regardless of content —
/// the failure mode that would make the idempotency test above pass for the wrong
/// reason. Gated on the shared GER40/USDJPY Sept-2024 archive; skips cleanly on a
/// data refusal.
#[test]
fn generalize_distinct_invocations_persist_distinct_campaign_documents() {
    const FROM_MS: &str = "1725148800000";
    const TO_MS: &str = "1727740799999";
    let (cwd, _g) = fresh_project();
    let fixture = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let invoke = |fast: &str| {
        let fast_axis = format!("sma_signal.fast.length={fast}");
        std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
            .args([
                "generalize", &fixture, "--real", "GER40,USDJPY",
                "--axis", &fast_axis, "--axis", "sma_signal.slow.length=12",
                "--stop-length", "14", "--stop-k", "2.0",
                "--from", FROM_MS, "--to", TO_MS,
            ])
            .current_dir(&cwd)
            .output()
            .expect("spawn aura")
    };
    let first = invoke("3");
    if first.status.code() == Some(1) {
        let stderr = String::from_utf8_lossy(&first.stderr);
        assert!(
            stderr.contains("no local data") || stderr.contains("no recorded geometry"),
            "exit 1 must be a data refusal, got: {stderr}"
        );
        eprintln!("skip: no local GER40/USDJPY data");
        return;
    }
    assert_eq!(first.status.code(), Some(0), "first run exit: {:?}", first.status);
    let second = invoke("4");
    assert_eq!(second.status.code(), Some(0), "second run exit: {:?}", second.status);

    let count = |sub: &str| {
        std::fs::read_dir(cwd.join("runs").join(sub))
            .map(|d| d.count())
            .unwrap_or(0)
    };
    assert_eq!(count("campaigns"), 2, "two distinct candidates persist two distinct campaign documents");
}

/// Property (#210 T3, dissolution; window semantics superseded by #213):
/// omitting `--from`/`--to` still completes — the dispatch rewrite's
/// window-resolution fallback (`dispatch_generalize` in main.rs) resolves ONE
/// shared campaign window from the INTERSECTION of the listed symbols' full
/// archive geometry and applies it to every instrument in the campaign
/// document (a `CampaignDoc` carries a single shared window, unlike the old
/// inline `run_generalize`, which let each instrument resolve its OWN
/// independent full window). A wrong intersection (or a panic on the
/// per-symbol probe) in that fallback would only surface when `--from`/`--to`
/// are absent — every other generalize e2e pins the explicit-window path,
/// leaving this one unexercised. The property is the no-window RESOLUTION
/// mechanics, not any particular archive's span, so it runs against
/// `fresh_project_with_data`'s tiny synthetic two-symbol archive (#250) —
/// hostless, no data-refusal skip arm needed.
#[test]
fn generalize_without_explicit_window_resolves_a_shared_window_and_completes() {
    let (cwd, _g) = fresh_project_with_data();
    let fixture = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args([
            "generalize", &fixture, "--real", "SYMA,SYMB",
            "--axis", "sma_signal.fast.length=3", "--axis", "sma_signal.slow.length=12",
            "--stop-length", "14", "--stop-k", "2.0",
        ])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura");
    assert_eq!(out.status.code(), Some(0), "exit: {:?}, stderr: {}", out.status, String::from_utf8_lossy(&out.stderr));
    let stdout = String::from_utf8(out.stdout).expect("utf-8 stdout");
    assert!(stdout.contains("\"generalize\":"), "aggregate object: {stdout}");
    assert!(stdout.contains("\"n_instruments\":2"), "two instruments graded: {stdout}");
    assert!(stdout.contains("\"family_id\":\"generalize-0\""), "family handle: {stdout}");
}

/// Property (#213, no-window semantics): invoked WITHOUT `--from`/`--to`, a
/// multi-symbol `generalize` resolves the ONE shared campaign window as the
/// INTERSECTION of the listed symbols' full archive windows — the latest start
/// and earliest end, i.e. the only span over which EVERY listed instrument
/// actually has data. `generalize` is the cross-instrument comparison (the
/// worst-case R floor, #146); a floor pooled over a different period per
/// instrument conflates the instrument axis with the period axis, so the common
/// window is the only one that isolates what the verb measures. Today's fallback
/// picks `symbols[0]`'s full window alone (the superseded shape test above):
/// listing the WIDER synthetic symbol first (`SYMA`, spanning 2024-01..08) keeps
/// SYMA's early start, whereas the intersection with `SYMB` (spanning 2024-03..06,
/// strictly inside SYMA's span) must start at SYMB's first bar and end at SYMB's
/// last — so on the current tree the resolved window differs from the
/// intersection on both bounds. Each symbol's full window is resolved
/// independently here: a single-symbol `sweep` with no window records it into its
/// generated campaign document, in the same Unix-ms currency and via the same
/// `campaign_window_ms(full_window)` probe generalize itself resolves against —
/// so the expected intersection is derived from the live archive, never a
/// hard-coded span. The archive's SPAN SHAPE is authored (#250,
/// `fresh_project_with_data`), not its size, so this runs hostless in
/// milliseconds instead of streaming two full host archives.
#[test]
fn generalize_without_explicit_window_resolves_the_intersection_of_all_symbols() {
    let (cwd, _g) = fresh_project_with_data();
    let fixture = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));

    // The resolved shared window persists verbatim in the generated campaign
    // document; read the (from_ms, to_ms) of the one whose `name` field matches.
    let window_of = |name: &str| -> (i64, i64) {
        let dir = cwd.join("runs").join("campaigns");
        for entry in std::fs::read_dir(&dir).expect("campaigns dir exists after a run") {
            let path = entry.expect("readable dir entry").path();
            let doc: serde_json::Value =
                serde_json::from_str(&std::fs::read_to_string(&path).expect("read campaign doc"))
                    .expect("campaign doc parses as JSON");
            if doc["name"].as_str() == Some(name) {
                let w = &doc["data"]["windows"][0];
                return (
                    w["from_ms"].as_i64().expect("from_ms is an integer"),
                    w["to_ms"].as_i64().expect("to_ms is an integer"),
                );
            }
        }
        panic!("no campaign document named {name:?} in {}", dir.display());
    };

    // Resolve one symbol's full archive window by running a single-symbol,
    // single-cell sweep with NO --from/--to; the window it records is exactly the
    // `campaign_window_ms(full_window)` value generalize resolves per symbol.
    let probe = |symbol: &str, name: &str| -> (i64, i64) {
        let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
            .args([
                "sweep", &fixture, "--real", symbol,
                "--axis", "sma_signal.fast.length=3", "--axis", "sma_signal.slow.length=12",
                "--name", name,
            ])
            .current_dir(&cwd)
            .output()
            .expect("spawn aura sweep");
        assert_eq!(
            out.status.code(), Some(0),
            "single-symbol window-probe sweep for {symbol} must run to exit 0: {}",
            String::from_utf8_lossy(&out.stderr)
        );
        window_of(name)
    };

    let syma = probe("SYMA", "probe-syma");
    let symb = probe("SYMB", "probe-symb");
    // Precondition: the archives genuinely differ, so "the intersection" is a
    // distinct claim from "symbols[0]'s full window" (else the test is vacuous).
    assert_ne!(
        syma, symb,
        "SYMA {syma:?} and SYMB {symb:?} must span different windows for the test to bite"
    );

    // The contract: the shared window is the intersection — latest start, earliest end.
    let expected = (syma.0.max(symb.0), syma.1.min(symb.1));

    // SYMA listed FIRST: today's `symbols[0]` fallback keeps SYMA's (wider)
    // window, which differs from the intersection on both bounds — so a pass here
    // can only mean the intersection semantics landed.
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args([
            "generalize", &fixture, "--real", "SYMA,SYMB",
            "--axis", "sma_signal.fast.length=3", "--axis", "sma_signal.slow.length=12",
            "--name", "gen-both",
        ])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura generalize");
    assert_eq!(
        out.status.code(), Some(0),
        "two-symbol no-window generalize must run to exit 0: {}",
        String::from_utf8_lossy(&out.stderr)
    );
    let resolved = window_of("gen-both");

    assert_eq!(
        resolved, expected,
        "no-window generalize must resolve the INTERSECTION {expected:?} of SYMA {syma:?} \
         and SYMB {symb:?} — not symbols[0]'s ({syma:?}) full window (resolved {resolved:?})"
    );
}

/// Property: a single-instrument generalize is refused at parse time (exit 2),
/// before any data access — so it asserts the arity refusal on any machine.
#[test]
fn generalize_refuses_a_single_instrument() {
    let fixture = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args([
            "generalize", &fixture, "--real", "GER40",
            "--axis", "sma_signal.fast.length=3", "--axis", "sma_signal.slow.length=12",
            "--stop-length", "14", "--stop-k", "2.0",
        ])
        .output()
        .expect("spawn aura");
    assert_eq!(out.status.code(), Some(2), "single instrument must exit 2");
}

/// Property: a non-R metric is refused before any instrument runs (the data-free
/// `check_r_metric` pre-check), so this asserts the R-only refusal on any machine.
#[test]
fn generalize_refuses_a_non_r_metric() {
    let fixture = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args([
            "generalize", &fixture, "--real", "GER40,USDJPY",
            "--axis", "sma_signal.fast.length=3", "--axis", "sma_signal.slow.length=12",
            "--stop-length", "14", "--stop-k", "2.0",
            "--metric", "total_pips",
        ])
        .output()
        .expect("spawn aura");
    assert_eq!(out.status.code(), Some(2), "a non-R metric must exit 2");
    let stderr = String::from_utf8_lossy(&out.stderr);
    assert!(stderr.contains("R-only"), "must name the R-only refusal, got: {stderr}");
}

/// Property: a duplicate instrument in the `--real` list is refused at the binary
/// boundary (exit 2, before any data access) — `GER40,GER40` never runs. The spec's
/// double-count hazard: a repeated symbol would be evaluated twice and so counted
/// twice in both the worst-case floor and the sign-agreement count, silently
/// corrupting the generalization grade. The parser unit test pins the grammar; this
/// pins that the parse error actually reaches `exit(2)` through the dispatch arm
/// (a swallowed Err would let a duplicated run proceed). Data-free — the duplicate
/// check precedes any geometry/archive lookup, so it asserts on any machine.
#[test]
fn generalize_refuses_a_duplicate_instrument() {
    let fixture = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args([
            "generalize", &fixture, "--real", "GER40,GER40",
            "--axis", "sma_signal.fast.length=3", "--axis", "sma_signal.slow.length=12",
            "--stop-length", "14", "--stop-k", "2.0",
        ])
        .output()
        .expect("spawn aura");
    assert_eq!(out.status.code(), Some(2), "a duplicate instrument must exit 2");
    assert!(
        out.stdout.is_empty(),
        "the refusal must not leak an aggregate to stdout, got: {:?}",
        String::from_utf8_lossy(&out.stdout)
    );
    let stderr = String::from_utf8_lossy(&out.stderr);
    assert!(stderr.contains("generalize"), "must surface the generalize usage, got: {stderr}");
}

/// Property: a multi-value axis is refused at the binary boundary (exit 2, no
/// run, data-free). A *candidate* is a single grid cell, not a sweep — so any
/// `--axis` carrying more than one value is refused, never silently widened
/// into a multi-cell run whose generalization grade would be ill-defined.
#[test]
fn generalize_refuses_a_multi_value_axis() {
    let fixture = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args([
            "generalize", &fixture, "--real", "GER40,USDJPY",
            "--axis", "sma_signal.fast.length=2,3", "--axis", "sma_signal.slow.length=12",
            "--stop-length", "14", "--stop-k", "2.0",
        ])
        .output()
        .expect("spawn aura");
    assert_eq!(out.status.code(), Some(2), "a multi-value axis must exit 2");
    let stderr = String::from_utf8_lossy(&out.stderr);
    assert!(stderr.contains("each --axis takes exactly one value"), "arity refusal: {stderr}");
    assert!(
        out.stdout.is_empty(),
        "the refusal must not leak an aggregate to stdout, got: {:?}",
        String::from_utf8_lossy(&out.stdout)
    );
}

/// Property: a `--axis` with no `--axis` flags at all is a usage error (exit 2,
/// no run) — generalize needs >= 1 axis to name a candidate, mirroring the sibling
/// sweep/walkforward "no axis" refusals. Data-free (fires before any archive access).
#[test]
fn generalize_refuses_no_axis() {
    let fixture = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args([
            "generalize", &fixture, "--real", "GER40,USDJPY",
            "--stop-length", "14", "--stop-k", "2.0",
        ])
        .output()
        .expect("spawn aura");
    assert_eq!(out.status.code(), Some(2), "zero axes must exit 2");
    let stderr = String::from_utf8_lossy(&out.stderr);
    assert!(stderr.contains("Usage: aura generalize"), "must print the generalize usage: {stderr}");
    assert!(
        out.stdout.is_empty(),
        "the refusal must not leak an aggregate to stdout, got: {:?}",
        String::from_utf8_lossy(&out.stdout)
    );
}

/// Property: an `--axis` name that is not among the loaded blueprint's sweepable
/// axes is refused before the raw-namespace strip or any archive access (exit 2),
/// echoing exactly the name the user typed — the same WRAPPED-probe preflight
/// `aura sweep` enforces (`aura_sweep_real_blueprint_refuses_a_raw_form_axis_name_before_data_access`).
#[test]
fn generalize_refuses_an_unknown_axis_name() {
    let fixture = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args([
            "generalize", &fixture, "--real", "GER40,USDJPY",
            "--axis", "sma_signal.nope=3", "--axis", "sma_signal.slow.length=12",
            "--stop-length", "14", "--stop-k", "2.0",
        ])
        .output()
        .expect("spawn aura");
    assert_eq!(out.status.code(), Some(2), "an unknown axis name must exit 2");
    let stderr = String::from_utf8_lossy(&out.stderr);
    assert!(
        stderr.contains("axis \"sma_signal.nope\"") && stderr.contains("sweepable axes"),
        "must name the unresolved axis exactly as typed: {stderr}"
    );
    assert!(
        out.stdout.is_empty(),
        "the refusal must not leak an aggregate to stdout, got: {:?}",
        String::from_utf8_lossy(&out.stdout)
    );
}

/// Property: a successful `aura generalize` persists its per-instrument runs as a
/// discoverable `CrossInstrument` family — the C12 comparison axis / C18 lineage at
/// the built-binary boundary. The unit `cli_families_persist_and_round_trip_per_kind`
/// covers Sweep/MonteCarlo/WalkForward but not CrossInstrument; this is the
/// cross-instrument sibling of `mc_runs_persists_a_monte_carlo_family_and_lists_it`:
/// after the run, `aura runs families` lists `generalize-0` with
/// `"kind":"CrossInstrument"` and `"members":2` (one member per instrument). A
/// regression that mis-tagged the family kind, or persisted the wrong member count,
/// would slip past the stdout-only happy-path assertion above but fail here. Gated on
/// local GER40/USDJPY data (the shared Sept-2024 window); skips cleanly when absent.
#[test]
fn generalize_persists_a_discoverable_cross_instrument_family() {
    const FROM_MS: &str = "1725148800000";
    const TO_MS: &str = "1727740799999";
    let (cwd, _g) = fresh_project();
    let fixture = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args([
            "generalize", &fixture, "--real", "GER40,USDJPY",
            "--axis", "sma_signal.fast.length=3", "--axis", "sma_signal.slow.length=12",
            "--stop-length", "14", "--stop-k", "2.0",
            "--from", FROM_MS, "--to", TO_MS,
        ])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura");
    if out.status.code() == Some(1) {
        let stderr = String::from_utf8_lossy(&out.stderr);
        assert!(
            stderr.contains("no local data") || stderr.contains("no recorded geometry"),
            "exit 1 must be a data refusal, got: {stderr}"
        );
        eprintln!("skip: no local GER40/USDJPY data");
        return;
    }
    assert_eq!(out.status.code(), Some(0), "exit: {:?}", out.status);

    // the run persisted the per-instrument members as a discoverable CrossInstrument family.
    let fams = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args(["runs", "families"])
        .current_dir(&cwd)
        .output()
        .expect("spawn families");
    assert!(fams.status.success(), "families exit: {:?}", fams.status);
    let fams_out = String::from_utf8(fams.stdout).expect("utf-8");
    // The dissolved path routes through the campaign executor, which persists a
    // per-instrument Sweep family per cell (each instrument's candidate run, now
    // a durable audit artifact) alongside the one CrossInstrument grade family
    // the sugar appends. Pin exactly one CrossInstrument family (the generalize
    // result) — not the total family count, which now includes those per-cell
    // Sweep families.
    let cross: Vec<&str> = fams_out
        .lines()
        .filter(|l| l.contains("\"kind\":\"CrossInstrument\""))
        .collect();
    assert_eq!(cross.len(), 1, "exactly one CrossInstrument family: {fams_out:?}");
    assert!(cross[0].contains("\"family_id\":\"generalize-0\""), "families: {fams_out:?}");
    assert!(cross[0].contains("\"members\":2"), "one member per instrument: {fams_out:?}");
}

/// Property (#210 T3, dissolution): the persisted `CrossInstrument` family's
/// members are attributed to the RIGHT instrument in the RIGHT cell order —
/// not merely correct in count. The dissolved path builds the family via
/// `cross_instrument_members`, which walks `outcome.cells` and trusts their
/// order/labelling, replacing the old `run_generalize`'s explicit
/// `zip(symbols, members)`. A cell-ordering or index bug in the new
/// extraction would still print the right aggregate (the campaign's own
/// `Generalize` stage computes that independently) while silently persisting
/// swapped or mislabelled members — a regression this test, not the exact-grade
/// anchor, would catch. Asserts `aura runs family generalize-0` lists GER40
/// first with the exact-grade anchor's GER40 `expectancy_r`, then USDJPY with
/// its own. Gated on the local GER40/USDJPY data (the shared Sept-2024 window).
#[test]
fn generalize_family_members_are_attributed_to_the_right_instrument() {
    const FROM_MS: &str = "1725148800000";
    const TO_MS: &str = "1727740799999";
    let (cwd, _g) = fresh_project();
    let fixture = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args([
            "generalize", &fixture, "--real", "GER40,USDJPY",
            "--axis", "sma_signal.fast.length=3", "--axis", "sma_signal.slow.length=12",
            "--stop-length", "14", "--stop-k", "2.0",
            "--from", FROM_MS, "--to", TO_MS,
        ])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura");
    if out.status.code() == Some(1) {
        let stderr = String::from_utf8_lossy(&out.stderr);
        assert!(
            stderr.contains("no local data") || stderr.contains("no recorded geometry"),
            "exit 1 must be a data refusal, got: {stderr}"
        );
        eprintln!("skip: no local GER40/USDJPY data");
        return;
    }
    assert_eq!(out.status.code(), Some(0), "exit: {:?}", out.status);

    let fam = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args(["runs", "family", "generalize-0"])
        .current_dir(&cwd)
        .output()
        .expect("spawn runs family");
    assert!(fam.status.success(), "runs family exit: {:?}", fam.status);
    let fam_out = String::from_utf8(fam.stdout).expect("utf-8");
    let members: Vec<serde_json::Value> = fam_out
        .lines()
        .filter(|l| l.starts_with('{'))
        .map(|l| serde_json::from_str(l).expect("member line parses as JSON"))
        .collect();
    assert_eq!(members.len(), 2, "one member per instrument: {fam_out:?}");

    assert_eq!(
        members[0]["manifest"]["instrument"].as_str(), Some("GER40"),
        "first member is GER40: {fam_out:?}"
    );
    assert_eq!(
        members[0]["metrics"]["r"]["expectancy_r"].as_f64(), Some(0.01056371324510624),
        "GER40's own expectancy_r, matching the exact-grade anchor: {fam_out:?}"
    );
    assert_eq!(
        members[1]["manifest"]["instrument"].as_str(), Some("USDJPY"),
        "second member is USDJPY: {fam_out:?}"
    );
    assert_eq!(
        members[1]["metrics"]["r"]["expectancy_r"].as_f64(), Some(0.005795903617609842),
        "USDJPY's own expectancy_r, matching the exact-grade anchor: {fam_out:?}"
    );
}

/// Property (#218 migration follow-up): `generalize`'s per-instrument members
/// route through the shared campaign path (`campaign_run.rs`'s
/// `run_blueprint_member` call, a code path distinct from sweep's in
/// `main.rs`), and that path ALSO stamps `manifest.project` with the running
/// project's provenance (C18/C16) for every persisted member — not just the
/// sweep chokepoint. Before the project-fixture migration this cli_run.rs
/// path only ever ran outside a project, so `project` was always `None` here.
/// Gated on the shared GER40/USDJPY Sept-2024 archive.
#[test]
fn generalize_family_members_stamp_project_provenance() {
    const FROM_MS: &str = "1725148800000";
    const TO_MS: &str = "1727740799999";
    let (cwd, _g) = fresh_project();
    let fixture = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args([
            "generalize", &fixture, "--real", "GER40,USDJPY",
            "--axis", "sma_signal.fast.length=3", "--axis", "sma_signal.slow.length=12",
            "--stop-length", "14", "--stop-k", "2.0",
            "--from", FROM_MS, "--to", TO_MS,
        ])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura");
    if out.status.code() == Some(1) {
        let stderr = String::from_utf8_lossy(&out.stderr);
        assert!(
            stderr.contains("no local data") || stderr.contains("no recorded geometry"),
            "exit 1 must be a data refusal, got: {stderr}"
        );
        eprintln!("skip: no local GER40/USDJPY data");
        return;
    }
    assert_eq!(out.status.code(), Some(0), "exit: {:?}", out.status);

    let fam = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args(["runs", "family", "generalize-0"])
        .current_dir(&cwd)
        .output()
        .expect("spawn runs family");
    assert!(fam.status.success(), "runs family exit: {:?}", fam.status);
    let fam_out = String::from_utf8(fam.stdout).expect("utf-8");
    let members: Vec<serde_json::Value> = fam_out
        .lines()
        .filter(|l| l.starts_with('{'))
        .map(|l| serde_json::from_str(l).expect("member line parses as JSON"))
        .collect();
    assert_eq!(members.len(), 2, "one member per instrument: {fam_out:?}");
    for m in &members {
        let project = &m["manifest"]["project"];
        assert_eq!(
            project["namespace"].as_str(), Some("demo"),
            "campaign-path member manifest carries the fixture project's namespace: {fam_out:?}"
        );
        let sha = project["dylib_sha256"].as_str()
            .unwrap_or_else(|| panic!("dylib_sha256 present: {fam_out:?}"));
        assert_eq!(sha.len(), 64, "dylib_sha256 is a full sha256 hex digest: {fam_out:?}");
    }
}

/// `aura sweep <blueprint.json> --axis <name>=<csv> …` (#166): sweeps a loaded CLOSED
/// blueprint over its named bound-override axes and persists a discoverable Sweep family.
/// Both bound knobs (fast/slow) must be re-opened by axes; the 2×2 grid yields 4 members.
#[test]
fn aura_sweep_loads_a_blueprint_and_persists_a_sweep_family() {
    let cwd = temp_cwd("blueprint-sweep");
    let fixture = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let out = Command::new(BIN)
        .args([
            "sweep", &fixture,
            "--axis", "sma_signal.fast.length=2,4",
            "--axis", "sma_signal.slow.length=8,16",
            "--name", "f",
        ])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura sweep blueprint");
    assert!(out.status.success(), "exit: {:?} stderr={}", out.status, String::from_utf8_lossy(&out.stderr));
    let fams = Command::new(BIN).args(["runs", "families"]).current_dir(&cwd).output().expect("spawn families");
    let fo = String::from_utf8(fams.stdout).expect("utf-8");
    assert!(fo.contains("\"kind\":\"Sweep\""), "families: {fo}");
    assert!(fo.contains("\"members\":4"), "2×2 grid -> four members: {fo}");
    let _ = std::fs::remove_dir_all(&cwd);
}

/// Property (#168, refuse-don't-lie): `aura sweep`/`aura walkforward` must REFUSE
/// `--trace` with a forward-pointer to the deferred per-member-trace feature (#224),
/// never silently accept it. The clap help ("also persists each member's taps") and
/// the chart `NotFound` hint ("run `aura sweep --trace` first") advertise
/// trace-recording, but no blueprint-mode arm persists per-member taps —
/// `run_blueprint_sweep` reads `let _ = persist;` and the dissolved `--real` sugar
/// sets `persist_taps: vec![]` — so a bare accept makes the surface lie. This mirrors
/// the existing `aura run` / `aura mc` `--trace` refusals (exit 2, refuse-don't-guess).
/// Driven over the SYNTHETIC path (both open knobs bound, no `--real`), so the test is
/// autonomous: no recorded archive, no project. Today `sweep --trace` exits 0 (silent
/// accept) and `walkforward --trace` exits 2 with a GENERIC usage string (no #224) —
/// both are the lying/pointer-less surface this pins.
#[test]
fn sweep_and_walkforward_refuse_trace_with_a_forward_pointer() {
    let fixture = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    for verb in ["sweep", "walkforward"] {
        let cwd = temp_cwd(&format!("{verb}-trace-refusal"));
        let out = Command::new(BIN)
            .args([
                verb,
                &fixture,
                "--axis",
                "sma_signal.fast.length=2,4",
                "--axis",
                "sma_signal.slow.length=8,16",
                "--trace",
                "fam",
            ])
            .current_dir(&cwd)
            .output()
            .expect("spawn aura <verb> --trace");
        assert_eq!(
            out.status.code(),
            Some(2),
            "`aura {verb} --trace` must be refused (exit 2); got {:?}, stderr: {}",
            out.status,
            String::from_utf8_lossy(&out.stderr),
        );
        let stderr = String::from_utf8_lossy(&out.stderr);
        assert!(
            stderr.contains("--trace"),
            "`aura {verb}` refusal names the offending flag: {stderr}"
        );
        assert!(
            stderr.contains("#224"),
            "`aura {verb}` refusal points forward to the deferred trace feature (#224): {stderr}"
        );
        assert!(
            out.stdout.is_empty(),
            "`aura {verb} --trace` must emit no report on stdout: {}",
            String::from_utf8_lossy(&out.stdout),
        );
        let _ = std::fs::remove_dir_all(&cwd);
    }
}

/// Property (#158/#164, C18/C11/C12): a blueprint sweep content-addresses its topology
/// — the canonical blueprint is persisted ONCE under `runs/blueprints/<topology_hash>.json`,
/// keyed by the very SHA256 every member's manifest carries, holding exactly the bytes
/// `blueprint_to_json(blueprint_from_json(doc))` yields. That stored topology is what makes
/// a generated sweep family reproducible from disk (the `aura reproduce` re-derivation).
#[test]
fn aura_sweep_persists_the_canonical_blueprint_keyed_by_topology_hash() {
    use aura_engine::{blueprint_from_json, blueprint_to_json};
    use aura_std::std_vocabulary;
    use sha2::{Digest, Sha256};

    let cwd = temp_cwd("blueprint-sweep-store");
    let fixture = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let out = Command::new(BIN)
        .args([
            "sweep", &fixture,
            "--axis", "sma_signal.fast.length=2,4",
            "--axis", "sma_signal.slow.length=8,16",
            "--name", "f",
        ])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura sweep blueprint");
    assert!(out.status.success(), "exit: {:?} stderr={}", out.status, String::from_utf8_lossy(&out.stderr));

    // exactly one blueprint stored for the whole family (one topology per family).
    let store = cwd.join("runs/blueprints");
    let mut entries: Vec<_> = std::fs::read_dir(&store)
        .unwrap_or_else(|e| panic!("the sweep must create the blueprint store {store:?}: {e}"))
        .map(|e| e.expect("dir entry").path())
        .collect();
    entries.sort();
    assert_eq!(entries.len(), 1, "one stored topology per family: {entries:?}");

    // keyed by a 64-hex topology hash, holding the canonical re-serialization.
    let stored_path = &entries[0];
    let stem = stored_path.file_stem().expect("stem").to_str().expect("utf-8");
    assert_eq!(stem.len(), 64, "content id is a 64-hex SHA256: {stem}");
    assert!(stem.bytes().all(|b| b.is_ascii_hexdigit()), "content id is hex: {stem}");

    let doc = std::fs::read_to_string(&fixture).expect("read fixture");
    let expected = blueprint_to_json(&blueprint_from_json(&doc, &|t| std_vocabulary(t)).expect("loads"))
        .expect("re-serializes");
    let stored = std::fs::read_to_string(stored_path).expect("read stored blueprint");
    assert_eq!(stored, expected, "the store holds the canonical blueprint bytes");

    // content-addressing, the feature's load-bearing property: the key IS the SHA256 of
    // the stored bytes — `get_blueprint(hash)` addresses exactly these bytes by identity,
    // not by coincidence of the store and the members computing the hash separately.
    let content_id: String =
        Sha256::digest(stored.as_bytes()).iter().map(|b| format!("{b:02x}")).collect();
    assert_eq!(stem, content_id, "the store key is the SHA256 of its content");

    // and that key is exactly what every family member's manifest carries — the hash the
    // reproduce path reads back to fetch the topology. All 4 members share the one topology.
    let members = std::fs::read_to_string(cwd.join("runs/families.jsonl")).expect("read family store");
    let carried = members.matches(&format!("\"topology_hash\":\"{stem}\"")).count();
    assert_eq!(carried, 4, "every member records the store key as its topology_hash: {members}");

    let _ = std::fs::remove_dir_all(&cwd);
}

/// An `--axis` name that does not resolve against the loaded blueprint's param_space
/// fails clean (a named BindError, non-zero exit), not a panic. The CLOSED fixture has
/// bound-override knobs, so a nonsense axis name (`nope`) is genuinely unknown — it names
/// neither a re-openable bound param nor an already-open knob (#246: a bound name would
/// re-open instead).
#[test]
fn aura_sweep_rejects_an_unknown_axis() {
    let fixture = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let out = Command::new(BIN)
        .args(["sweep", &fixture, "--axis", "nope=1,2", "--name", "f"])
        .output()
        .expect("spawn aura sweep unknown-axis");
    assert_ne!(out.status.code(), Some(0), "an unknown axis must fail the sweep");
    let stderr = String::from_utf8(out.stderr).expect("utf-8");
    assert!(
        stderr.contains("Knob") || stderr.to_lowercase().contains("nope"),
        "names the unresolved axis: {stderr}"
    );
}

/// Property (#158, C18 "re-derives full results on demand"): `aura reproduce <family-id>`
/// re-derives every member of a persisted sweep family from the content-addressed
/// blueprint store and reports each bit-identical, exiting 0. Drives the verb through the
/// binary — the `reproduce_family` stdout contract (a per-member `bit-identical` line plus
/// a `reproduced N/N` summary) that the `reproduce_family_in` unit seam bypasses.
#[test]
fn aura_reproduce_re_derives_a_persisted_sweep_family() {
    let cwd = temp_cwd("blueprint-reproduce");
    let fixture = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let sweep = Command::new(BIN)
        .args([
            "sweep", &fixture,
            "--axis", "sma_signal.fast.length=2,4",
            "--axis", "sma_signal.slow.length=8,16",
            "--name", "f",
        ])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura sweep blueprint");
    assert!(sweep.status.success(), "sweep exit: {:?} stderr={}", sweep.status, String::from_utf8_lossy(&sweep.stderr));

    // `--trace f` names the family `f-0`; reproduce it from disk.
    let out = Command::new(BIN)
        .args(["reproduce", "f-0"])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura reproduce");
    assert!(out.status.success(), "reproduce exit: {:?} stderr={}", out.status, String::from_utf8_lossy(&out.stderr));
    let stdout = String::from_utf8(out.stdout).expect("utf-8 stdout");
    // one verdict line per member (all bit-identical) + the summary; the label renders
    // param values portably (`length=2`, not the `I64(2)` Debug form).
    assert_eq!(
        stdout.lines().filter(|l| l.contains("reproduced: bit-identical")).count(),
        4,
        "every member re-derives bit-identically: {stdout}"
    );
    assert!(stdout.contains("sma_signal.fast.length=2"), "label renders values portably: {stdout}");
    assert!(!stdout.contains("I64("), "label must not leak the Scalar Debug form: {stdout}");
    assert!(stdout.contains("reproduced 4/4 members bit-identically"), "summary line: {stdout}");
    let _ = std::fs::remove_dir_all(&cwd);
}

/// Property (C1/C18 integrity net): when a persisted member's stored metrics no longer
/// match its re-derivation, `aura reproduce` reports that member `DIVERGED` and exits 1
/// — the non-happy branch of the verb. Forced by tampering one member's on-disk
/// `total_pips` after the sweep (the store's bit-identical compare IS the integrity check).
#[test]
fn aura_reproduce_reports_a_diverged_member_and_exits_one() {
    let cwd = temp_cwd("blueprint-reproduce-diverge");
    let fixture = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let sweep = Command::new(BIN)
        .args([
            "sweep", &fixture,
            "--axis", "sma_signal.fast.length=2,4",
            "--axis", "sma_signal.slow.length=8,16",
            "--name", "f",
        ])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura sweep blueprint");
    assert!(sweep.status.success(), "sweep exit: {:?} stderr={}", sweep.status, String::from_utf8_lossy(&sweep.stderr));

    // corrupt the first member's stored `total_pips` so its re-run metrics differ.
    let store = cwd.join("runs/families.jsonl");
    let s = std::fs::read_to_string(&store).expect("read family store");
    let key = "\"total_pips\":";
    let at = s.find(key).expect("a member records total_pips") + key.len();
    let end = at + s[at..].find(',').expect("total_pips value terminates");
    let corrupted = format!("{}999.0{}", &s[..at], &s[end..]);
    std::fs::write(&store, corrupted).expect("write corrupted store");

    let out = Command::new(BIN)
        .args(["reproduce", "f-0"])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura reproduce");
    assert_eq!(out.status.code(), Some(1), "a diverged member exits 1: stderr={}", String::from_utf8_lossy(&out.stderr));
    let stdout = String::from_utf8(out.stdout).expect("utf-8 stdout");
    assert!(stdout.contains("reproduced: DIVERGED"), "the tampered member is reported DIVERGED: {stdout}");
    assert!(stdout.contains("reproduced 3/4 members bit-identically"), "summary counts the diverged member: {stdout}");
    let _ = std::fs::remove_dir_all(&cwd);
}

/// E2E (acc 2): persist a blueprint sweep, then `aura reproduce <id>` re-derives every
/// member bit-identically from the content-addressed store and exits 0.
#[test]
fn aura_reproduce_re_derives_a_persisted_sweep_bit_identically() {
    let cwd = temp_cwd("reproduce_roundtrip");
    let fixture = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));

    let sweep = std::process::Command::new(BIN)
        .args([
            "sweep",
            &fixture,
            "--axis",
            "sma_signal.fast.length=2",
            "--axis",
            "sma_signal.slow.length=4,6",
            "--name",
            "smacross",
        ])
        .current_dir(&cwd)
        .output()
        .expect("run aura sweep");
    assert!(sweep.status.success(), "sweep ok: {}", String::from_utf8_lossy(&sweep.stderr));

    let repro = std::process::Command::new(BIN)
        .args(["reproduce", "smacross-0"])
        .current_dir(&cwd)
        .output()
        .expect("run aura reproduce");
    let stdout = String::from_utf8_lossy(&repro.stdout);
    assert!(repro.status.success(), "reproduce exits 0: {}", String::from_utf8_lossy(&repro.stderr));
    assert!(
        stdout.contains("reproduced 2/2 members bit-identically"),
        "every member re-derives: {stdout}"
    );

    let _ = std::fs::remove_dir_all(&cwd);
}

/// E2E (acc 1+3+4): `aura mc <closed blueprint.json> --seeds N` builds a MonteCarlo
/// family (one member per seed), writes exactly one content-addressed blueprint, and
/// `aura reproduce <id>` re-derives every member bit-identically (exit 0).
#[test]
fn aura_mc_over_a_blueprint_reproduces_bit_identically() {
    let cwd = temp_cwd("mc-blueprint-reproduce");
    let fixture = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));

    let mc = Command::new(BIN)
        .args(["mc", &fixture, "--seeds", "4", "--name", "mcx"])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura mc blueprint");
    assert!(mc.status.success(), "mc exit: {:?} stderr={}", mc.status, String::from_utf8_lossy(&mc.stderr));
    let mc_out = String::from_utf8(mc.stdout).expect("utf-8 stdout");
    // one member line per seed (each carries "seed":n) + one aggregate line.
    assert_eq!(
        mc_out.lines().filter(|l| l.contains("\"seed\":")).count(),
        4,
        "one member line per seed: {mc_out}"
    );
    assert!(mc_out.contains("mc_aggregate"), "prints the aggregate line: {mc_out}");

    // exactly one blueprint stored for the whole family (one topology per family).
    let store = cwd.join("runs/blueprints");
    let entries: Vec<_> = std::fs::read_dir(&store)
        .unwrap_or_else(|e| panic!("mc must create the blueprint store {store:?}: {e}"))
        .map(|e| e.expect("dir entry").path())
        .collect();
    assert_eq!(entries.len(), 1, "one stored topology per family: {entries:?}");

    let repro = Command::new(BIN)
        .args(["reproduce", "mcx-0"])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura reproduce");
    assert!(repro.status.success(), "reproduce exit: {:?} stderr={}", repro.status, String::from_utf8_lossy(&repro.stderr));
    let repro_out = String::from_utf8(repro.stdout).expect("utf-8 stdout");
    assert!(
        repro_out.contains("reproduced 4/4 members bit-identically"),
        "every MC member re-derives: {repro_out}"
    );
    let _ = std::fs::remove_dir_all(&cwd);
}

/// E2E (#173): an IS-refit walk-forward over a loaded blueprint persists a
/// content-addressed WalkForward family that `aura reproduce`s bit-identically —
/// each OOS member's windowed slice + winner params are recovered from its manifest.
#[test]
fn aura_walkforward_over_a_blueprint_reproduces_bit_identically() {
    let cwd = temp_cwd("blueprint-walkforward-reproduce");
    let bp = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let run = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args(["walkforward", &bp, "--axis", "sma_signal.fast.length=2,3",
               "--axis", "sma_signal.slow.length=4,6", "--name", "wfr"])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura walkforward");
    assert_eq!(run.status.code(), Some(0), "stderr: {}", String::from_utf8_lossy(&run.stderr));
    let repro = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args(["reproduce", "wfr-0"])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura reproduce");
    assert_eq!(repro.status.code(), Some(0));
    let out = String::from_utf8(repro.stdout).expect("utf-8");
    assert!(out.contains("reproduced 3/3 members bit-identically"), "stdout: {out}");
    let _ = std::fs::remove_dir_all(&cwd);
}

/// E2E (#173): the loaded-blueprint walk-forward runs + persists a WalkForward
/// family (3 OOS members) with a {"walkforward":…} summary carrying oos_r, and
/// stores exactly one content-addressed blueprint.
#[test]
fn aura_walkforward_over_a_blueprint_persists_a_family() {
    let cwd = temp_cwd("blueprint-walkforward-persist");
    let bp = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args(["walkforward", &bp, "--axis", "sma_signal.fast.length=2,3",
               "--axis", "sma_signal.slow.length=4,6", "--name", "wfx"])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura walkforward");
    assert_eq!(out.status.code(), Some(0), "stderr: {}", String::from_utf8_lossy(&out.stderr));
    let stdout = String::from_utf8(out.stdout).expect("utf-8");
    assert_eq!(stdout.matches("wfx-").count(), 3, "3 OOS member lines: {stdout}");
    assert!(stdout.contains("\"walkforward\""), "summary line: {stdout}");
    assert!(stdout.contains("\"oos_r\""), "R-measured summary: {stdout}");
    let families = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args(["runs", "families"]).current_dir(&cwd).output().expect("spawn families");
    let fo = String::from_utf8(families.stdout).expect("utf-8");
    assert!(fo.contains("\"kind\":\"WalkForward\""), "families: {fo}");
    let n = std::fs::read_dir(cwd.join("runs").join("blueprints")).map(|d| d.count()).unwrap_or(0);
    assert_eq!(n, 1, "exactly one content-addressed blueprint stored");
    let _ = std::fs::remove_dir_all(&cwd);
}

/// E2E (#173): a CLOSED blueprint with no --axis (nothing to re-fit) is a usage
/// error (exit 2), naming that >= 1 --axis is required.
#[test]
fn aura_walkforward_over_a_blueprint_rejects_no_axis() {
    let bp = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args(["walkforward", &bp])
        .output()
        .expect("spawn aura walkforward (no axis)");
    assert_eq!(out.status.code(), Some(2));
    let stderr = String::from_utf8(out.stderr).expect("utf-8");
    assert!(stderr.contains("requires >= 1 --axis"), "stderr: {stderr}");
}

/// E2E (#173/#184, safety): a malformed blueprint is rejected at the dispatch
/// boundary (exit 2) rather than panicking, phrased house-style (#184's
/// `blueprint_load_prose` convention) — never the raw `UnknownNodeType` Debug
/// leak (c0110 fieldtest finding, mirroring `aura_run_rejects_an_unknown_node_blueprint`).
#[test]
fn aura_walkforward_over_a_blueprint_rejects_malformed() {
    let bp = format!("{}/tests/fixtures/unknown_node.json", env!("CARGO_MANIFEST_DIR"));
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args(["walkforward", &bp, "--axis", "sma_signal.fast.length=2,3"])
        .output()
        .expect("spawn aura walkforward (malformed)");
    assert_eq!(out.status.code(), Some(2), "malformed must be rejected, not panic");
    let stderr = String::from_utf8(out.stderr).expect("utf-8");
    assert!(stderr.contains(r#"unknown node type "Nope""#), "house-style prose names the type: {stderr}");
    assert!(!stderr.contains("UnknownNodeType"), "does not leak the Debug variant name: {stderr}");
    assert!(!stderr.contains("panicked"), "must reject cleanly: {stderr}");
}

/// E2E (#173, panic-safety): a `--axis` naming a knob that is not in the loaded
/// blueprint's param_space fails clean (named BindError, exit 2, no panic). This
/// pins the in-closure fallible-bind path — `blueprint_sweep_over` returns a
/// `BindError` from inside the `walk_forward` window closure — which is a DISTINCT
/// branch from the dispatch-boundary blueprint-parse rejection (the malformed test):
/// the blueprint here is well-formed and passes dispatch, then the per-window
/// re-fit's bind fails. A regression to `.expect()` on that bind would panic here
/// while leaving the malformed test green.
#[test]
fn aura_walkforward_over_a_blueprint_rejects_an_unknown_axis() {
    let bp = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args(["walkforward", &bp, "--axis", "nope=1,2"])
        .output()
        .expect("spawn aura walkforward (unknown axis)");
    assert_eq!(out.status.code(), Some(2), "an unknown axis must fail clean, not panic");
    let stderr = String::from_utf8(out.stderr).expect("utf-8");
    assert!(
        stderr.contains("names no param") && stderr.contains("nope"),
        "names the unresolved axis via the prose payload (#269): {stderr}"
    );
    assert!(!stderr.contains("panicked"), "must reject cleanly, never panic: {stderr}");
}

/// E2E (#177): a rejected `--axis` on a loaded walk-forward blueprint is reported
/// EXACTLY ONCE, however many IS/OOS windows the roll would have spanned — mirroring
/// the sibling `aura sweep` / `aura mc` paths, which validate the axis ONCE up front
/// and emit their rejection a single time. The property this protects: axis
/// resolution is a single pre-flight, not a per-window re-raise.
///
/// The CLOSED fixture has an empty param_space, so ANY `--axis` name is an
/// `UnknownKnob`. The pre-fix code validates the axis INSIDE the per-window closure
/// that `walk_forward` fans out in parallel across the windows, so more than one
/// window `eprintln!`s the same rejection before the racing `exit(2)` tears the
/// process down — a race whose emit count is 1 or 2 per run. A single invocation
/// would therefore be a flaky assertion (it emits once ~1/3 of the time by luck), so
/// this drives the verb repeatedly: observing the double-emit at least once is then
/// near-certain, and the assertion is that EVERY run emits the rejection exactly
/// once. The fix (hoist axis resolution to one pre-flight before the windowed
/// fan-out) makes every run deterministically single.
#[test]
fn aura_walkforward_emits_an_unknown_axis_rejection_once() {
    let bp = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    for attempt in 1..=20 {
        let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
            .args(["walkforward", &bp, "--axis", "graph.fast.length=2,3"])
            .output()
            .expect("spawn aura walkforward (unknown axis)");
        assert_eq!(out.status.code(), Some(2), "an unknown axis must fail clean, not panic");
        let stderr = String::from_utf8(out.stderr).expect("utf-8");
        assert!(!stderr.contains("panicked"), "must reject cleanly, never panic: {stderr}");
        let emits = stderr.matches("names no param").count();
        assert_eq!(
            emits, 1,
            "attempt {attempt}: the axis rejection must be emitted exactly once, got {emits}: {stderr:?}",
        );
    }
}

/// Property (#269): the walkforward verb's unknown-axis rejection reaches a
/// user's stderr as prose — the same register `aura sweep`'s boundary errors
/// adopted in #247 — never wrapped in the `UnknownKnob("…")` Rust `Debug`
/// frame. The walkforward path already carries a fully-prosed payload INSIDE
/// that wrapper (`axis <name>: names no param of this blueprint … — see …
/// --list-axes`), so the fix sheds the frame rather than authoring new prose:
/// the message that reaches stderr is exactly that payload, the axis named as
/// typed, exit still 2. Data-free — the axis is rejected at the resolution
/// pre-flight before any window runs, so no archive and no project are needed.
///
/// NB: this is the post-#269 contract. The two sibling tests above
/// (`aura_walkforward_over_a_blueprint_rejects_an_unknown_axis`,
/// `aura_walkforward_emits_an_unknown_axis_rejection_once`) still pin the
/// `UnknownKnob` substring — stale pins of the defective frame that the GREEN
/// slice repoints onto the prose fragment.
#[test]
fn walkforward_unknown_axis_rejection_renders_as_prose_not_a_debug_frame() {
    let bp = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args(["walkforward", &bp, "--axis", "nope=1,2"])
        .output()
        .expect("spawn aura walkforward (unknown axis)");
    assert_eq!(
        out.status.code(),
        Some(2),
        "an unknown axis stays a usage refusal (exit 2, not a panic): status={:?}",
        out.status
    );
    let stderr = String::from_utf8(out.stderr).expect("utf-8");
    // The headline — the whole defect: no Rust `Debug` frame reaches the user.
    assert!(
        !stderr.contains("UnknownKnob"),
        "must not leak the `UnknownKnob(…)` Debug frame around the prose: {stderr}"
    );
    // The prose payload survives intact — the axis named as typed, the
    // names-no-param wording, and the `--list-axes` pointer (the sweep
    // sibling's register).
    assert!(
        stderr.contains("nope"),
        "the prose names the offending axis exactly as typed: {stderr}"
    );
    assert!(
        stderr.contains("names no param"),
        "the prose carries the boundary payload wording: {stderr}"
    );
    assert!(
        stderr.contains("--list-axes"),
        "the prose points at the discovery surface, like the sweep sibling: {stderr}"
    );
    assert!(
        !stderr.contains("panicked"),
        "must reject cleanly, never panic: {stderr}"
    );
}

/// E2E (acc 2): `aura mc <open blueprint.json>` is rejected with a named error + exit 2
/// (NOT a panic / exit 101) — MC binds no axis, so a free knob has no binder.
#[test]
fn aura_mc_rejects_an_open_blueprint() {
    let cwd = temp_cwd("mc-blueprint-open-reject");
    let fixture = format!("{}/tests/fixtures/r_sma_open.json", env!("CARGO_MANIFEST_DIR"));
    let out = Command::new(BIN)
        .args(["mc", &fixture, "--seeds", "4"])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura mc open-blueprint");
    assert_eq!(out.status.code(), Some(2), "an open blueprint fails clean (exit 2, not a panic): stderr={}", String::from_utf8_lossy(&out.stderr));
    let stderr = String::from_utf8(out.stderr).expect("utf-8");
    assert!(stderr.contains("closed blueprint"), "names the closed-blueprint requirement: {stderr}");
    let _ = std::fs::remove_dir_all(&cwd);
}

/// E2E (#176): `aura run <open blueprint.json>` refuses an open blueprint the same way the
/// sibling `aura mc` does — a clean named error + exit 2, NEVER a `compile_with_params`
/// arity panic (exit 101). The `run` dispatch bootstraps over the EMPTY point (a
/// closed-blueprint assumption), so a free knob must be rejected at the dispatch boundary
/// before that bootstrap, mirroring `blueprint_mc_family`'s closed-guard.
#[test]
fn aura_run_rejects_an_open_blueprint_without_panicking() {
    let cwd = temp_cwd("run-blueprint-open-reject");
    let fixture = format!("{}/tests/fixtures/r_sma_open.json", env!("CARGO_MANIFEST_DIR"));
    let out = Command::new(BIN)
        .args(["run", &fixture])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura run open-blueprint");
    let stderr = String::from_utf8_lossy(&out.stderr).into_owned();
    assert_eq!(
        out.status.code(),
        Some(2),
        "an open blueprint fails clean (exit 2, not a panic/exit 101): stderr={stderr}"
    );
    assert!(
        stderr.contains("closed blueprint"),
        "names the closed-blueprint requirement (mirroring `aura mc`): {stderr}"
    );
    assert!(
        !stderr.contains("panicked"),
        "must refuse at the dispatch boundary, never panic: {stderr}"
    );
    let _ = std::fs::remove_dir_all(&cwd);
}

/// E2E (negative): `aura reproduce <unknown>` is a hard error (exit non-zero + named
/// cause on stderr), distinct from `runs family <id>`'s treat-as-empty exit 0.
#[test]
fn aura_reproduce_rejects_an_unknown_family() {
    let cwd = temp_cwd("reproduce_unknown");
    let out = std::process::Command::new(BIN)
        .args(["reproduce", "no-such-family-1"])
        .current_dir(&cwd)
        .output()
        .expect("run aura reproduce");
    assert!(!out.status.success(), "unknown family exits non-zero");
    assert!(
        String::from_utf8_lossy(&out.stderr).contains("no such family 'no-such-family-1'"),
        "names the cause: {}",
        String::from_utf8_lossy(&out.stderr)
    );
    let _ = std::fs::remove_dir_all(&cwd);
}

// Note (#159 cut 4 collateral): this removes
// `aura_reproduce_rejects_a_non_generated_family` — its fixture was a
// HARD-WIRED (bare `sweep --name`) family with no `topology_hash`; sweep is
// blueprint-only now and every blueprint-sweep member stamps a
// `topology_hash`, so no surviving CLI surface can construct the fixture.
// The defensive "not a generated run" refusal in `main.rs` stays as dead-path
// safety for a malformed/legacy `families.jsonl`. Removed as collateral, not
// repointed.

/// E2E (refuse-don't-guess): `aura reproduce` when the content-addressed blueprint was
/// removed from the store exits non-zero, naming the missing blueprint — never re-runs
/// against a guessed topology (C10/C18).
#[test]
fn aura_reproduce_rejects_a_missing_stored_blueprint() {
    let cwd = temp_cwd("reproduce_missing_store");
    let fixture = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let sweep = std::process::Command::new(BIN)
        .args([
            "sweep",
            &fixture,
            "--axis",
            "sma_signal.fast.length=2",
            "--axis",
            "sma_signal.slow.length=4",
            "--name",
            "bp",
        ])
        .current_dir(&cwd)
        .output()
        .expect("run aura sweep");
    assert!(sweep.status.success(), "blueprint sweep ok: {}", String::from_utf8_lossy(&sweep.stderr));
    std::fs::remove_dir_all(cwd.join("runs").join("blueprints")).expect("remove the blueprint store");
    let out = std::process::Command::new(BIN)
        .args(["reproduce", "bp-0"])
        .current_dir(&cwd)
        .output()
        .expect("run aura reproduce");
    assert!(!out.status.success(), "a missing stored blueprint exits non-zero");
    assert!(
        String::from_utf8_lossy(&out.stderr).contains("missing from store"),
        "names the cause: {}",
        String::from_utf8_lossy(&out.stderr)
    );
    let _ = std::fs::remove_dir_all(&cwd);
}

/// E2E (#169/#246): `aura sweep <open blueprint.json> --list-axes` prints one
/// `<name>:<kind>` line per open sweepable knob, in `param_space()` order,
/// followed by one `<name>:<kind> default=<value>` line per BOUND param (the
/// fixture's `bias.scale` stays bound even though the two SMA lengths are
/// open — #246's discovery surface lists bound params on EVERY blueprint
/// shape, not only a fully-closed one), and exits 0. The names are prefixed
/// by the wrapping (sma_signal.*) — exactly the strings `--axis` binds — so a
/// user can discover then sweep or re-open without guessing.
#[test]
fn aura_sweep_list_axes_prints_prefixed_names() {
    let bp = format!("{}/tests/fixtures/r_sma_open.json", env!("CARGO_MANIFEST_DIR"));
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args(["sweep", &bp, "--list-axes"])
        .output()
        .expect("spawn aura sweep --list-axes");
    assert_eq!(out.status.code(), Some(0));
    let stdout = String::from_utf8(out.stdout).expect("utf-8");
    assert_eq!(
        stdout,
        "sma_signal.fast.length:I64\n\
         sma_signal.slow.length:I64\n\
         sma_signal.bias.scale:F64 default=0.5\n"
    );
}

/// E2E (#246): `--list-axes` on a CLOSED blueprint (all knobs bound) prints one
/// `<bp>.<node>.<param>:<KIND> default=<value>` line per bound param — the
/// bound value IS the default, overridable by naming the same path as an
/// `--axis` (#246's re-open contract) — and exits 0. The sibling pin above
/// (`aura_sweep_list_axes_prints_prefixed_names`) proves the same bound-param
/// lines also appear on a PARTIALLY open blueprint, after its open lines.
#[test]
fn aura_sweep_list_axes_closed_prints_bound_defaults() {
    let bp = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args(["sweep", &bp, "--list-axes"])
        .output()
        .expect("spawn aura sweep --list-axes (closed)");
    assert_eq!(out.status.code(), Some(0));
    let stdout = String::from_utf8(out.stdout).expect("utf-8");
    assert_eq!(
        stdout,
        "sma_signal.fast.length:I64 default=2\n\
         sma_signal.slow.length:I64 default=4\n\
         sma_signal.bias.scale:F64 default=0.5\n"
    );
}

/// E2E (#169): `--list-axes` is a standalone query — combining it with a real
/// sweep flag (`--axis`) is a usage error (exit 2), naming that it lists axes
/// and takes no other flags. A query cannot also seed a sweep.
#[test]
fn aura_sweep_list_axes_rejects_other_flags() {
    let bp = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args(["sweep", &bp, "--list-axes", "--axis", "sma_signal.fast.length=2,3"])
        .output()
        .expect("spawn aura sweep --list-axes + --axis");
    assert_eq!(out.status.code(), Some(2));
    let stderr = String::from_utf8(out.stderr).expect("utf-8");
    assert!(stderr.contains("--list-axes lists axes"), "stderr was: {stderr}");
}

/// E2E (#169/#184, safety): `--list-axes` over a malformed blueprint rejects
/// cleanly (exit 2), phrased house-style — never the raw `UnknownNodeType`
/// Debug leak (c0110 fieldtest finding) — rather than panicking; the axis
/// probe reloads the doc, so the dispatch-boundary validation must run first,
/// never a raw `.expect`.
#[test]
fn aura_sweep_list_axes_rejects_malformed_blueprint() {
    let bp = format!("{}/tests/fixtures/unknown_node.json", env!("CARGO_MANIFEST_DIR"));
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args(["sweep", &bp, "--list-axes"])
        .output()
        .expect("spawn aura sweep --list-axes (malformed)");
    assert_eq!(out.status.code(), Some(2), "malformed blueprint must be rejected, not panic");
    let stderr = String::from_utf8(out.stderr).expect("utf-8");
    assert!(stderr.contains(r#"unknown node type "Nope""#), "house-style prose names the type: {stderr}");
    assert!(!stderr.contains("UnknownNodeType"), "does not leak the Debug variant name: {stderr}");
    assert!(!stderr.contains("panicked"), "must reject cleanly, not panic: {stderr}");
}

/// E2E (#210 c0110 fieldtest, "aura sweep <op-script> rejects the op-script
/// array with a raw serde error"): `aura sweep`'s blueprint slot accepts only
/// the built blueprint envelope; an un-built op-script (a JSON array) is
/// shape-discriminated and refused with a targeted, house-style hint — never
/// the leaked `Json(Error("invalid type: map, expected u32", ...))`.
#[test]
fn aura_sweep_rejects_an_op_script_array_with_a_build_first_hint() {
    let dir = temp_cwd("sweep_rejects_op_script");
    let op_script = dir.join("ops.json");
    std::fs::write(&op_script, r#"[{"op":"source","role":"price","kind":"F64"}]"#)
        .expect("write op-script fixture");
    let out = Command::new(BIN)
        .args(["sweep", op_script.to_str().unwrap(), "--list-axes"])
        .output()
        .expect("spawn aura sweep (op-script)");
    assert_eq!(out.status.code(), Some(2), "an op-script array must be refused, not panic");
    let stderr = String::from_utf8_lossy(&out.stderr).into_owned();
    assert!(
        stderr.contains("op-script") && stderr.contains("aura graph build"),
        "names the op-script cause + the fix: {stderr}"
    );
    assert!(!stderr.contains("Json(Error"), "does not leak the raw serde error: {stderr}");
    let _ = std::fs::remove_dir_all(&dir);
}

/// E2E (#210 c0110 fieldtest, house-style loader prose): a genuinely malformed
/// (syntactically invalid) JSON document in the sweep blueprint slot is
/// refused with `blueprint_load_prose`'s wording, never the raw `{e:?}` Debug
/// form the dispatch used to print.
#[test]
fn aura_sweep_rejects_malformed_json_house_style() {
    let dir = temp_cwd("sweep_rejects_malformed_json");
    let bad = dir.join("bad.json");
    std::fs::write(&bad, "{not valid json").expect("write malformed fixture");
    let out = Command::new(BIN)
        .args(["sweep", bad.to_str().unwrap(), "--list-axes"])
        .output()
        .expect("spawn aura sweep (malformed json)");
    assert_eq!(out.status.code(), Some(2), "malformed JSON must be refused, not panic");
    let stderr = String::from_utf8_lossy(&out.stderr).into_owned();
    assert!(
        stderr.contains("blueprint document is not valid JSON"),
        "house-style prose, not a raw Debug leak: {stderr}"
    );
    assert!(!stderr.contains("Json(Error"), "does not leak the raw serde Debug form: {stderr}");
    let _ = std::fs::remove_dir_all(&dir);
}

/// E2E (#169, the load-bearing "listed == swept" invariant): a name emitted by
/// `--list-axes` is accepted VERBATIM by `--axis` — the discovered axis namespace
/// and the swept axis namespace are the same set (shared-probe construction). The
/// axis names here are DERIVED from the `--list-axes` output, never hardcoded, then
/// fed straight back into a real sweep: the sweep exits 0 and persists a `Sweep`
/// family, proving every listed name resolved (no UnknownKnob). This is the
/// discovery-removes-guessing property #169 exists to close — were the list and
/// sweep surfaces ever to diverge, a discovered name would be rejected here and this
/// round-trip would fail, whereas the literal-pinned per-surface tests could not see it.
#[test]
fn aura_sweep_list_axes_names_round_trip_into_a_working_sweep() {
    let cwd = temp_cwd("blueprint-list-axes-roundtrip");
    let bp = format!("{}/tests/fixtures/r_sma_open.json", env!("CARGO_MANIFEST_DIR"));

    // 1) discover the axis names + kinds (do NOT hardcode them) — this is what a user
    //    reads. #246: the fixture now lists a BOUND param (`bias.scale`, F64) alongside
    //    the two open I64 lengths, so the CSV below must be typed per discovered kind,
    //    not assumed integer.
    let list = Command::new(BIN)
        .args(["sweep", &bp, "--list-axes"])
        .output()
        .expect("spawn aura sweep --list-axes");
    assert_eq!(list.status.code(), Some(0));
    let listed = String::from_utf8(list.stdout).expect("utf-8");
    let names: Vec<(String, String)> = listed
        .lines()
        .map(|l| {
            let mut parts = l.splitn(2, ':');
            let name = parts.next().expect("a `name:kind` line").to_string();
            let kind = parts
                .next()
                .expect("kind after ':'")
                .split_whitespace()
                .next()
                .expect("a kind token")
                .to_string();
            (name, kind)
        })
        .collect();
    assert!(!names.is_empty(), "the open fixture must list >= 1 axis");

    // 2) feed the DISCOVERED names straight back into a real sweep. Values are
    //    position-assigned (first axis small, rest large) to stay in a non-degenerate
    //    SMA regime without re-hardcoding the names; each axis takes two values, typed
    //    per its discovered kind so a bound F64 default (e.g. `bias.scale`) resolves
    //    instead of tripping `KindMismatch` on an integer-shaped token.
    let mut args: Vec<String> = vec!["sweep".into(), bp.clone()];
    for (i, (name, kind)) in names.iter().enumerate() {
        let vals: &str = match (i == 0, kind.as_str()) {
            (true, "F64") => "2.0,4.0",
            (true, _) => "2,4",
            (false, "F64") => "8.0,16.0",
            (false, _) => "8,16",
        };
        args.push("--axis".into());
        args.push(format!("{name}={vals}"));
    }
    let sweep = Command::new(BIN)
        .args(&args)
        .current_dir(&cwd)
        .output()
        .expect("spawn aura sweep over discovered axes");
    assert!(
        sweep.status.success(),
        "every listed name must be a valid --axis; exit={:?} stderr={}",
        sweep.status,
        String::from_utf8_lossy(&sweep.stderr)
    );

    // 3) the sweep persisted a Sweep family whose grid spans exactly the discovered
    //    axes (each bound to two values -> 2^k members) — the names resolved end-to-end.
    let fams = Command::new(BIN).args(["runs", "families"]).current_dir(&cwd).output().expect("spawn families");
    let fo = String::from_utf8(fams.stdout).expect("utf-8");
    assert!(fo.contains("\"kind\":\"Sweep\""), "families: {fo}");
    let expected = 1usize << names.len(); // two values per discovered axis
    assert!(fo.contains(&format!("\"members\":{expected}")), "grid over the discovered axes: {fo}");

    let _ = std::fs::remove_dir_all(&cwd);
}

/// `--version` / `-V` surface the crate version on stdout and exit 0 (GNU MUST):
/// a version query is a successful query, not a usage error. The expected string
/// is the crate version itself (`CARGO_PKG_VERSION`, i.e. the workspace version),
/// so this stays honest across version bumps and keeps one source of truth.
#[test]
fn version_flag_prints_version_to_stdout_and_exits_zero() {
    for flag in ["--version", "-V"] {
        let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
            .arg(flag)
            .output()
            .expect("spawn aura");
        assert_eq!(out.status.code(), Some(0), "{flag}: exit {:?}", out.status);
        let stdout = String::from_utf8_lossy(&out.stdout);
        assert!(stdout.contains(env!("CARGO_PKG_VERSION")), "{flag} stdout: {stdout:?}");
    }
}

/// `--version` carries the ENGINE COMMIT beside the semver, drawn from the SAME
/// `AURA_COMMIT` build-time provenance the run manifest stamps (the
/// `option_env!("AURA_COMMIT")` that fills `RunManifest.commit`), so a stale
/// binary is distinguishable at the CLI surface *before* any run (#266). Shape:
/// `aura 0.1.0 (<commit-ish>)`, exit 0.
///
/// Property, not a specific sha: the expected commit is read from the same
/// provenance source the built binary was compiled with — this integration test
/// receives aura-cli's build-script `rustc-env`, so it sees the identical value
/// (dirty/`unknown` suffixes included) rather than today's HEAD, keeping the
/// assertion honest against a lagging or unclean tree. The load-bearing claim is
/// that the parenthesized commit rides beside the still-present semver.
#[test]
fn version_flag_carries_the_engine_commit_from_manifest_provenance() {
    // The single source of truth — the same env the manifest's commit is sourced
    // from; falls back to "unknown" identically to `sim_optimal_manifest`.
    let commit = option_env!("AURA_COMMIT").unwrap_or("unknown");
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .arg("--version")
        .output()
        .expect("spawn aura");
    assert_eq!(out.status.code(), Some(0), "--version exit {:?}", out.status);
    let stdout = String::from_utf8_lossy(&out.stdout);
    // Semver stays present (one source of truth, CARGO_PKG_VERSION).
    assert!(
        stdout.contains(env!("CARGO_PKG_VERSION")),
        "--version still prints the semver: {stdout:?}"
    );
    // ...and the engine commit rides beside it, parenthesized, from the shared
    // manifest provenance — the load-bearing assertion #266 adds.
    assert!(
        stdout.contains(&format!("({commit})")),
        "--version prints the engine commit ({commit}) beside the semver: {stdout:?}"
    );
}

/// Subcommand help is SCOPED: `aura sweep --help` names sweep's own flags and is
/// NOT byte-identical to `aura run --help` — each subcommand documents its own
/// options rather than reprinting one shared global blob (the #131 uniform-help
/// surface retired).
#[test]
fn subcommand_help_is_scoped_not_uniform() {
    let sweep = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args(["sweep", "--help"]).output().expect("spawn");
    let run = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args(["run", "--help"]).output().expect("spawn");
    assert_eq!(sweep.status.code(), Some(0), "sweep --help exit");
    assert_eq!(run.status.code(), Some(0), "run --help exit");
    let sweep_out = String::from_utf8_lossy(&sweep.stdout);
    let run_out = String::from_utf8_lossy(&run.stdout);
    assert!(sweep_out.contains("--strategy"), "sweep help names its flags: {sweep_out:?}");
    assert_ne!(sweep_out, run_out, "per-subcommand help must be scoped, not uniform");
}

/// The GNU `--flag=value` equals form is accepted, equivalent to the
/// space-separated `--flag value`. (#159 cut 4: `--harness` retired along with
/// the built-in default harness; `--seed` on a shipped closed blueprint is the
/// surviving vehicle flag for this clap-grammar property.)
#[test]
fn gnu_equals_form_is_accepted() {
    let fixture = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args(["run", &fixture, "--seed=0"]).output().expect("spawn");
    assert_eq!(out.status.code(), Some(0),
        "run --seed=0 should run; stderr: {}", String::from_utf8_lossy(&out.stderr));
}

/// The `--` end-of-options terminator is recognized (a bare `--` after the flags
/// is a no-op that still runs). (#159 cut 4: repointed off the retired
/// `--harness` onto `--seed` over a shipped closed blueprint.)
#[test]
fn double_dash_terminator_is_recognized() {
    let fixture = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args(["run", &fixture, "--seed", "0", "--"]).output().expect("spawn");
    assert_eq!(out.status.code(), Some(0),
        "run ... -- should run; stderr: {}", String::from_utf8_lossy(&out.stderr));
}

/// GNU long-option abbreviation (spec acceptance criterion 4): an unambiguous prefix
/// of a long flag is accepted (`--se` → `--seed`), via clap `infer_long_args`
/// on the root command (which propagates to subcommands). An LLM/automation caller
/// always writes the full flag, but GNU compliance (the cycle's ratified purpose)
/// includes abbreviation, and clap delivers it with one root attribute. (#159 cut 4:
/// repointed off the retired `--harness` onto `--seed` over a shipped closed
/// blueprint.)
#[test]
fn long_option_abbreviation_is_accepted() {
    let fixture = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args(["run", &fixture, "--se", "0"]).output().expect("spawn");
    assert_eq!(out.status.code(), Some(0),
        "run --se 0 (abbrev of --seed) should run; stderr: {}",
        String::from_utf8_lossy(&out.stderr));
}

/// Property (#175, dual-grammar built-in branch stays strict): a leading positional
/// that is NOT an existing `.json` file is a usage error (exit 2), never silently
/// swallowed — for every dual-grammar subcommand (run/sweep/walkforward/mc). The
/// clap migration models the loaded-blueprint `[BLUEPRINT]` as an OPTIONAL positional
/// on one shared args struct; that makes a typo'd leading token (`aura run bogus`,
/// meant as `--harness bogus`) structurally parseable where the old hand-parser
/// rejected any stray token. Each built-in handler therefore re-asserts the guard
/// (`if a.blueprint.is_some() { usage; exit 2 }`). Dropping any one guard would
/// silently run the default harness on a mistyped invocation — the exact regression
/// clap's optional positional invites and this pins shut. Exit 2 preserved (the
/// usage/runtime split is iteration 2); nothing on stdout on the refusal path.
#[test]
fn dual_grammar_stray_positional_is_a_usage_error_not_swallowed() {
    // A trailing flag is added where the built-in grammar needs one to reach the
    // guard rather than a bare-run default; `bogus` never ends in `.json`, so the
    // is_file() discriminator sends every case down the built-in branch.
    for argv in [
        &["run", "bogus"][..],
        &["sweep", "bogus", "--name", "x"][..],
        &["walkforward", "bogus"][..],
        &["mc", "bogus"][..],
    ] {
        let out = Command::new(BIN).args(argv).output().expect("spawn aura <sub> <stray-positional>");
        assert_eq!(
            out.status.code(),
            Some(2),
            "{argv:?} must be refused (exit 2), got {:?}; stderr: {}",
            out.status,
            String::from_utf8_lossy(&out.stderr)
        );
        assert!(
            out.stdout.is_empty(),
            "{argv:?} must not emit a report on stdout: {:?}",
            String::from_utf8_lossy(&out.stdout)
        );
    }
}

/// Property (#175, dual-grammar discriminator keys on `is_file()`, not the suffix):
/// `aura run <name>.json` where the file does NOT exist is refused with the same
/// usage error as a bare, blueprint-less `run` (exit 2), NOT read as a blueprint.
/// The discriminator is `ends_with(".json") && Path::is_file()`; a regression that
/// dropped the `is_file()` conjunct (suffix-only) would send this down the
/// blueprint-read branch and fail with a file-read error instead — a different,
/// misleading refusal. (#159 cut 4: the built-in `--harness` branch retired, so the
/// no-blueprint arm now emits the loaded-blueprint usage line unconditionally; the
/// assertion is repointed onto that line, still distinct from a file-read error.)
/// The positive `.json`-file path stays covered by
/// `aura_run_loads_and_runs_a_blueprint_file`.
#[test]
fn dual_grammar_discriminator_requires_an_existing_file_not_just_json_suffix() {
    let out = Command::new(BIN)
        .args(["run", "definitely-not-a-real-file.json"])
        .output()
        .expect("spawn aura run <nonexistent>.json");
    assert_eq!(
        out.status.code(),
        Some(2),
        "a .json name that is not a file must be refused (exit 2), got {:?}; stderr: {}",
        out.status,
        String::from_utf8_lossy(&out.stderr)
    );
    assert!(out.stdout.is_empty(), "no report on stdout: {:?}", String::from_utf8_lossy(&out.stdout));
    let stderr = String::from_utf8_lossy(&out.stderr);
    assert!(
        stderr.contains("Usage: aura run <blueprint.json>"),
        "the no-blueprint usage arm (not the blueprint-read path) must handle it: {stderr:?}"
    );
}

/// The exit-code partition (#175 iteration 2, deviation #8): a USAGE error (a
/// malformed command line) exits 2; a RUNTIME failure (a well-formed command
/// whose needed data/state is missing) exits 1. Pins the partition as a property.
/// (#159 cut 4: `run` is blueprint-only now, so the runtime-failure arm needs a
/// shipped closed blueprint to clear the usage grammar before the no-geometry
/// refusal fires.)
#[test]
fn exit_codes_partition_usage_two_from_runtime_one() {
    // usage: an unknown flag is a command-line error → exit 2
    let usage = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args(["run", "--bogus"]).output().expect("spawn");
    assert_eq!(usage.status.code(), Some(2),
        "unknown flag is a usage error → 2; stderr: {}", String::from_utf8_lossy(&usage.stderr));
    // runtime: a well-formed command whose symbol has no recorded data → exit 1
    let fixture = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let runtime = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args(["run", &fixture, "--real", "NONEXISTENT"]).output().expect("spawn");
    assert_eq!(runtime.status.code(), Some(1),
        "no data for a valid command is a runtime failure → 1; stderr: {}",
        String::from_utf8_lossy(&runtime.stderr));
}

/// Characterization pin (byte-identity anchor for the mc R-bootstrap dissolution,
/// #210, the last verb; argv since migrated to the generic form by #220).
/// `aura mc <blueprint> --real --axis …` pools the per-window OOS trade-R
/// series and r-bootstraps E[R] through the campaign
/// `[std::sweep(argmax), std::walk_forward, std::monte_carlo]`
/// process, whose terminal monte_carlo stage does the
/// `StageBootstrap::PooledOos(r_bootstrap(...))` seeded from the campaign seed -- so
/// the campaign seed carries the mc `--seed`. The wf winners are
/// deflation-seed-independent (argmax), so the pooled series, and thus the
/// bootstrap, is stable across that seed remap (the walkforward anchor already
/// proved winner seed-independence). The multi-point grid makes the per-window
/// winner selection non-degenerate. The pinned bootstrap grade of this fixed 2025
/// GER40 invocation ORIGINATES from the pre-#210 inline rolling walk-forward
/// (`mc_r_bootstrap_report`) under the retired welded `aura mc --strategy r-sma
/// --real` grammar, and survived both the campaign dissolution and the #220 argv
/// migration byte-identically (the acceptance gate then, the regression anchor
/// now). Gated on the GER40 archive; skips
/// cleanly on a data refusal.
#[test]
fn mc_r_bootstrap_real_e2e_pins_the_exact_current_grade() {
    const FROM_MS: &str = "1735689600000";
    const TO_MS: &str = "1767225599000";
    let (cwd, _g) = fresh_project();
    let fixture = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args([
            "mc", &fixture, "--real", "GER40",
            "--axis", "sma_signal.fast.length=3,5", "--axis", "sma_signal.slow.length=12,20",
            "--stop-length", "14", "--stop-k", "2.0",
            "--block-len", "5", "--resamples", "1000", "--seed", "42",
            "--from", FROM_MS, "--to", TO_MS,
        ])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura");
    if out.status.code() == Some(1) {
        let stderr = String::from_utf8_lossy(&out.stderr);
        assert!(
            stderr.contains("no local data") || stderr.contains("no recorded geometry"),
            "exit 1 must be a data refusal, got: {stderr}"
        );
        eprintln!("skip: no local GER40 data");
        return;
    }
    assert_eq!(out.status.code(), Some(0), "exit: {:?}", out.status);
    let stdout = String::from_utf8(out.stdout).expect("utf-8 stdout");
    let grade_line = stdout
        .lines()
        .find(|l| l.starts_with("{\"mc_r_bootstrap\":"))
        .unwrap_or_else(|| panic!("the mc_r_bootstrap grade line: {stdout}"));
    let v: serde_json::Value = serde_json::from_str(grade_line).expect("grade line parses as JSON");
    let mc = &v["mc_r_bootstrap"];
    assert_eq!(mc["block_len"].as_u64(), Some(5), "block_len: {grade_line}");
    assert_eq!(mc["n_resamples"].as_u64(), Some(1000), "n_resamples: {grade_line}");
    assert_eq!(mc["n_trades"].as_u64(), Some(20681), "pooled OOS trade count: {grade_line}");
    // EXACT bootstrap floats -- the byte-identity anchor the dissolution must preserve.
    assert_eq!(mc["e_r"]["mean"].as_f64(), Some(-0.0025753095301594307), "E[R] mean: {grade_line}");
    assert_eq!(mc["e_r"]["p50"].as_f64(), Some(-0.0023049902245975227), "E[R] median: {grade_line}");
    assert_eq!(mc["prob_le_zero"].as_f64(), Some(0.558), "P(E[R] <= 0): {grade_line}");
}

/// Property (#220 mc vertical): `aura mc --real` R-bootstraps ANY sweepable
/// blueprint through the campaign path, not just the historical r-sma candidate
/// — the whole point of dropping `--strategy r-sma`/`--fast`/`--slow` for a
/// positional blueprint + generic `--axis`. Every other mc E2E test here still
/// exercises `r_sma.json`; a regression that silently special-cased the
/// pipeline's axis lookup back to a hardcoded `sma_signal.*` namespace would
/// ship green there while breaking every other blueprint. This runs the
/// unrelated r-breakout candidate (the ganged `channel_length` axis, no
/// `sma_signal` node in sight) end-to-end and asserts a well-formed
/// `mc_r_bootstrap` grade: the argv-echoed knobs (`block_len`/`n_resamples`)
/// pin exactly, `n_trades`/`e_r.mean`/`prob_le_zero` are asserted present and
/// well-typed (their exact values are strategy-specific, already the r-sma
/// pin's job above). Gated on the shared GER40 archive; skips cleanly on a
/// data refusal.
#[test]
fn mc_dissolves_a_non_r_sma_blueprint() {
    const FROM_MS: &str = "1735689600000";
    const TO_MS: &str = "1767225599000";
    let (cwd, _g) = fresh_project();
    let fixture = format!("{}/tests/fixtures/r_breakout_open.json", env!("CARGO_MANIFEST_DIR"));
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args([
            "mc", &fixture, "--real", "GER40",
            "--axis", "r_breakout_signal.channel_length=10,20",
            "--stop-length", "14", "--stop-k", "2.0",
            "--block-len", "5", "--resamples", "1000", "--seed", "42",
            "--from", FROM_MS, "--to", TO_MS,
        ])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura");
    if out.status.code() == Some(1) {
        let stderr = String::from_utf8_lossy(&out.stderr);
        assert!(
            stderr.contains("no local data") || stderr.contains("no recorded geometry"),
            "exit 1 must be a data refusal, got: {stderr}"
        );
        eprintln!("skip: no local GER40 data");
        return;
    }
    assert_eq!(out.status.code(), Some(0), "exit: {:?}", out.status);
    let stdout = String::from_utf8(out.stdout).expect("utf-8 stdout");
    let grade_line = stdout
        .lines()
        .find(|l| l.starts_with("{\"mc_r_bootstrap\":"))
        .unwrap_or_else(|| panic!("the mc_r_bootstrap grade line: {stdout}"));
    let v: serde_json::Value = serde_json::from_str(grade_line).expect("grade line parses as JSON");
    let mc = &v["mc_r_bootstrap"];
    assert_eq!(mc["block_len"].as_u64(), Some(5), "block_len (argv echo): {grade_line}");
    assert_eq!(mc["n_resamples"].as_u64(), Some(1000), "n_resamples (argv echo): {grade_line}");
    assert!(mc["n_trades"].as_u64().is_some_and(|n| n > 0), "pooled OOS trade count present: {grade_line}");
    assert!(mc["e_r"]["mean"].as_f64().is_some_and(|m| m.is_finite()), "E[R] mean present + finite: {grade_line}");
    assert!(
        mc["prob_le_zero"].as_f64().is_some_and(|p| (0.0..=1.0).contains(&p)),
        "P(E[R] <= 0) present + in [0,1]: {grade_line}"
    );
}

/// Property: real-data `aura mc <blueprint> --real --axis …` (the #210 dissolution
/// of the retired welded `--strategy r-sma` form, argv since migrated by #220) is
/// thin sugar over the one campaign
/// path — a successful run durably auto-registers exactly one generated process
/// document, one generated campaign document (carrying the constant "mc" name and the
/// stop as a non-empty single risk regime), and one campaign-run record, and emits the
/// single `mc_r_bootstrap` grade line. The pipeline's walk_forward stage persists
/// the one family; the leading `std::grid` stage only enumerates (#256) and the
/// terminal monte_carlo stage is an annotator — neither adds a family. This is
/// the observable proof the inline path is gone
/// and the dissolution runs through the executor. Gated on the GER40 2025 archive;
/// skips cleanly on a data refusal.
#[test]
fn mc_dissolves_through_the_campaign_path() {
    const FROM_MS: &str = "1735689600000";
    const TO_MS: &str = "1767225599000";
    let (cwd, _g) = fresh_project();
    let fixture = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args([
            "mc", &fixture, "--real", "GER40",
            "--axis", "sma_signal.fast.length=3,5", "--axis", "sma_signal.slow.length=12,20",
            "--stop-length", "14", "--stop-k", "2.0",
            "--block-len", "5", "--resamples", "1000", "--seed", "42",
            "--from", FROM_MS, "--to", TO_MS,
        ])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura");
    if out.status.code() == Some(1) {
        let stderr = String::from_utf8_lossy(&out.stderr);
        assert!(
            stderr.contains("no local data") || stderr.contains("no recorded geometry"),
            "exit 1 must be a data refusal, got: {stderr}"
        );
        eprintln!("skip: no local GER40 data");
        return;
    }
    assert_eq!(out.status.code(), Some(0), "exit: {:?}", out.status);

    // the single grade line is emitted (and nothing else on stdout).
    let stdout = String::from_utf8(out.stdout).expect("utf-8 stdout");
    assert_eq!(stdout.lines().count(), 1, "mc prints exactly one grade line: {stdout:?}");
    assert!(
        stdout.lines().next().is_some_and(|l| l.starts_with("{\"mc_r_bootstrap\":")),
        "the one line is the mc_r_bootstrap grade: {stdout}"
    );

    let count = |sub: &str| {
        std::fs::read_dir(cwd.join("runs").join(sub))
            .map(|d| d.count())
            .unwrap_or(0)
    };
    assert_eq!(count("processes"), 1, "one generated process document registered");
    assert_eq!(count("campaigns"), 1, "one generated campaign document registered");
    let runs_log = std::fs::read_to_string(cwd.join("runs").join("campaign_runs.jsonl"))
        .expect("campaign_runs.jsonl exists after a sugar run");
    assert_eq!(runs_log.lines().count(), 1, "one campaign run recorded");

    let campaigns_dir = cwd.join("runs").join("campaigns");
    let campaign_doc_path = std::fs::read_dir(&campaigns_dir)
        .expect("campaigns dir exists")
        .next()
        .expect("exactly one campaign document")
        .expect("readable dir entry")
        .path();
    let campaign_doc_json = std::fs::read_to_string(&campaign_doc_path).expect("read campaign doc");
    assert!(
        campaign_doc_json.contains("\"name\":\"mc\""),
        "the generated campaign document carries the constant mc name: {campaign_doc_json}"
    );
    assert!(
        campaign_doc_json.contains("\"risk\":[")
            && campaign_doc_json.contains("\"length\":14")
            && campaign_doc_json.contains("\"k\":2.0"),
        "the stop rides a non-empty risk regime: {campaign_doc_json}"
    );

    let fams = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args(["runs", "families"])
        .current_dir(&cwd)
        .output()
        .expect("spawn families");
    let fams_out = String::from_utf8_lossy(&fams.stdout).into_owned();
    assert_eq!(
        fams_out.lines().filter(|l| l.contains("\"kind\":\"Sweep\"")).count(),
        0,
        "the enumerate-only std::grid leading stage persists no Sweep family: {fams_out}"
    );
    assert_eq!(
        fams_out.lines().filter(|l| l.contains("\"kind\":\"WalkForward\"")).count(),
        1,
        "one WalkForward family holding the per-window OOS reports: {fams_out}"
    );
    assert_eq!(
        fams_out.lines().filter(|l| l.contains("\"kind\":\"MonteCarlo\"")).count(),
        0,
        "the terminal monte_carlo stage is an annotator, not a family: {fams_out}"
    );
}

/// Property (#239): a real-data `aura mc --real` invocation over a window SHORTER
/// than the fixed 90-day-in-sample / 30-day-out-of-sample roller runs the R-bootstrap
/// to completion by fitting the injected walk-forward roller down to the campaign
/// window, instead of refusing late at the executor with "walk_forward windows do not
/// fit the campaign window" (exit 1, no remedy reachable from the mc surface). The
/// short window is a FIXED ~30 days (2025-04-16..2025-05-16, the tail of the
/// ~135-day span the walkforward e2e tests use): what the property needs is "a
/// data-carrying window far shorter than the roller", which a fixed recent window
/// satisfies without the full-archive probe sweep a live-span derivation costs —
/// the archive only grows forward, so the window never falls off its end. The
/// leading std::grid stage only enumerates (nothing runs over the short window); the
/// refusal fires at stage 1 (the walk_forward roller), so this bites only when the
/// window genuinely has data. Gated on the GER40 archive; skips cleanly on a data-less host.
#[test]
fn mc_real_fits_the_injected_roller_to_a_short_window() {
    if !local_data_present() {
        eprintln!("skip: no local data at {}", data_server::DEFAULT_DATA_PATH);
        return;
    }
    let (cwd, _g) = fresh_project();
    let fixture = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));

    // ~30 days, far shorter than the fixed 90+30-day roller, so the injected
    // walk_forward cannot fit unless it scales down to the window.
    const SUB_FROM_MS: &str = "1744761600000";
    const SUB_TO_MS: &str = "1747353600000";

    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args([
            "mc", &fixture, "--real", "GER40",
            "--axis", "sma_signal.fast.length=3,5", "--axis", "sma_signal.slow.length=12,20",
            "--stop-length", "14", "--stop-k", "2.0",
            "--block-len", "5", "--resamples", "100", "--seed", "42",
            "--from", SUB_FROM_MS, "--to", SUB_TO_MS,
        ])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura mc");

    let stderr = String::from_utf8_lossy(&out.stderr);
    // Defensive: a genuine data refusal (never the late window-fit refusal this test
    // pins) still means "no usable data on this host" — skip, don't fail.
    if out.status.code() == Some(1)
        && (stderr.contains("no local data") || stderr.contains("no recorded geometry"))
    {
        eprintln!("skip: no local GER40 data");
        return;
    }
    assert_eq!(
        out.status.code(), Some(0),
        "a short real window must fit the injected roller and run to completion, \
         not refuse late; stderr: {stderr}"
    );
    let stdout = String::from_utf8(out.stdout).expect("utf-8 stdout");
    assert!(
        stdout.lines().any(|l| l.starts_with("{\"mc_r_bootstrap\":")),
        "the mc_r_bootstrap grade line must be emitted for the fitted short window: {stdout}"
    );
}

/// Fork A: the stop is a single risk regime on the dissolved mc path — a multi-value
/// `--stop-length`/`--stop-k` is a usage refusal (exit 2), not a swept axis. Mirrors
/// `walkforward_dissolved_refuses_a_multi_value_stop`. NOT gated — the refusal is pure
/// argument parsing, before any data access.
#[test]
fn mc_dissolved_refuses_a_multi_value_stop() {
    let cwd = temp_cwd("mc-multi-stop");
    let fixture = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args([
            "mc", &fixture, "--real", "GER40",
            "--axis", "sma_signal.fast.length=3", "--axis", "sma_signal.slow.length=12",
            "--stop-length", "14,20", "--stop-k", "2.0",
        ])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura");
    assert_eq!(out.status.code(), Some(2), "multi-value stop is a usage refusal");
    let stderr = String::from_utf8_lossy(&out.stderr);
    assert!(stderr.contains("single risk regime"), "stop-regime refusal: {stderr}");
}

/// Property: an `--axis` name that is not among the loaded blueprint's sweepable
/// axes is refused before the raw-namespace strip or any archive access (exit 2),
/// echoing exactly the name the user typed — mc shares the identical WRAPPED-probe
/// preflight `walkforward_dissolved_refuses_an_unknown_axis_name` pins for
/// walkforward (both dispatchers now call the same `validate_and_register_axes`
/// helper); this test pins mc's own copy so a future edit cannot silently break it
/// while walkforward's stays green. Data-free: no archive/store access.
#[test]
fn mc_dissolved_refuses_an_unknown_axis_name() {
    let cwd = temp_cwd("mc-unknown-axis");
    let fixture = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args([
            "mc", &fixture, "--real", "GER40",
            "--axis", "sma_signal.nope=3", "--axis", "sma_signal.slow.length=12",
            "--stop-length", "14", "--stop-k", "2.0",
        ])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura");
    assert_eq!(out.status.code(), Some(2), "an unknown axis name must exit 2");
    let stderr = String::from_utf8_lossy(&out.stderr);
    assert!(
        stderr.contains("axis \"sma_signal.nope\"") && stderr.contains("sweepable axes"),
        "must name the unresolved axis exactly as typed: {stderr}"
    );
    assert!(
        out.stdout.is_empty(),
        "the refusal must not leak an aggregate to stdout, got: {:?}",
        String::from_utf8_lossy(&out.stdout)
    );
    assert!(!cwd.join("runs").exists(), "a refused axis name must not start a real run");
}

/// Property: the one load-bearing decision new to the mc dissolution — `translate_mc`
/// maps `campaign.seed` to the mc `--seed` (unlike `translate_walkforward`, which
/// hardcodes `seed: 0`) — actually flows through the executor to the emitted grade,
/// not just into the generated document's content id (already unit-pinned by
/// `translate_mc_diverging_seeds_and_stops_do_not_collide_content_ids`). Two runs that
/// differ ONLY in `--seed` must emit a different `e_r` distribution (the bootstrap
/// resample draw is seeded) while pooling the IDENTICAL `n_trades` (the wf winners are
/// argmax-selected, hence deflation-seed-independent — the walkforward anchor already
/// proved this for the shared roller): the seed perturbs the resample, never the
/// underlying OOS trade-R series. Without this test, a dropped seed remap (e.g. mc
/// always bootstrapping at a fixed internal seed) would still pass every other mc
/// dissolution test, since none of them compare two distinct `--seed` runs. Gated on
/// the GER40 2025 archive; skips cleanly on a data refusal.
#[test]
fn mc_dissolved_seed_changes_the_bootstrap_draw_not_the_pooled_series() {
    const FROM_MS: &str = "1735689600000";
    const TO_MS: &str = "1767225599000";
    let (cwd, _g) = fresh_project();
    let fixture = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let run = |seed: &str| {
        std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
            .args([
                "mc", &fixture, "--real", "GER40",
                "--axis", "sma_signal.fast.length=3,5", "--axis", "sma_signal.slow.length=12,20",
                "--stop-length", "14", "--stop-k", "2.0",
                "--block-len", "5", "--resamples", "1000", "--seed", seed,
                "--from", FROM_MS, "--to", TO_MS,
            ])
            .current_dir(&cwd)
            .output()
            .expect("spawn aura")
    };
    let out42 = run("42");
    if out42.status.code() == Some(1) {
        let stderr = String::from_utf8_lossy(&out42.stderr);
        assert!(
            stderr.contains("no local data") || stderr.contains("no recorded geometry"),
            "exit 1 must be a data refusal, got: {stderr}"
        );
        eprintln!("skip: no local GER40 data");
        return;
    }
    let out7 = run("7");
    assert_eq!(out42.status.code(), Some(0), "exit (seed 42): {:?}", out42.status);
    assert_eq!(out7.status.code(), Some(0), "exit (seed 7): {:?}", out7.status);

    let parse = |out: &std::process::Output| -> serde_json::Value {
        let stdout = String::from_utf8_lossy(&out.stdout).into_owned();
        let line = stdout
            .lines()
            .find(|l| l.starts_with("{\"mc_r_bootstrap\":"))
            .unwrap_or_else(|| panic!("the mc_r_bootstrap grade line: {stdout}"))
            .to_string();
        serde_json::from_str(&line).expect("grade line parses as JSON")
    };
    let v42 = parse(&out42);
    let v7 = parse(&out7);

    assert_eq!(
        v42["mc_r_bootstrap"]["n_trades"], v7["mc_r_bootstrap"]["n_trades"],
        "the pooled OOS trade-R series is seed-independent (argmax winners): {v42} vs {v7}"
    );
    assert_ne!(
        v42["mc_r_bootstrap"]["e_r"]["mean"], v7["mc_r_bootstrap"]["e_r"]["mean"],
        "a different --seed must move the resampled E[R] mean: {v42} vs {v7}"
    );
}

/// #220 weld-gone proof: the walkforward campaign path runs an arbitrary user
/// blueprint with axis names the CLI has never heard of — the r-sma weld
/// (hardcoded fast/slow) is structurally gone. Gated on the GER40 archive;
/// skips cleanly when absent.
#[test]
fn walkforward_campaign_runs_an_arbitrary_blueprint() {
    if !local_data_present() {
        eprintln!("skip: no local data at {}", data_server::DEFAULT_DATA_PATH);
        return;
    }
    const FROM_MS: &str = "1735689600000";
    const TO_MS: &str = "1767225599000";
    let (cwd, _g) = fresh_project();
    let fixture = format!("{}/tests/fixtures/r_breakout_open.json", env!("CARGO_MANIFEST_DIR"));
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args([
            "walkforward", &fixture, "--real", "GER40",
            "--axis", "r_breakout_signal.channel_length=10,20",
            "--stop-length", "3", "--stop-k", "2.0",
            "--from", FROM_MS, "--to", TO_MS,
        ])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura");
    if out.status.code() == Some(1) {
        let stderr = String::from_utf8_lossy(&out.stderr);
        assert!(
            stderr.contains("no local data") || stderr.contains("no recorded geometry"),
            "exit 1 must be a data refusal, got: {stderr}"
        );
        eprintln!("skip: no local GER40 data");
        return;
    }
    assert_eq!(
        out.status.code(),
        Some(0),
        "exit: {:?} stderr: {}",
        out.status,
        String::from_utf8_lossy(&out.stderr)
    );
    let stdout = String::from_utf8(out.stdout).expect("utf-8 stdout");
    let grade_line = stdout
        .lines()
        .find(|l| l.starts_with("{\"walkforward\":"))
        .unwrap_or_else(|| panic!("the walkforward summary line: {stdout}"));
    let v: serde_json::Value = serde_json::from_str(grade_line).expect("summary line parses as JSON");
    let wf = &v["walkforward"];
    assert!(wf["windows"].as_u64().is_some_and(|w| w >= 1), "at least one window: {grade_line}");
    // param_stability rows: channel_length, stop_length, stop_k.
    let ps = wf["param_stability"].as_array().expect("param_stability is an array");
    assert_eq!(ps.len(), 3, "one ganged channel axis + the stop pair: {grade_line}");
}

/// #220 weld-gone proof for mc: the R-bootstrap pipeline runs the same
/// arbitrary breakout blueprint. Gated on the GER40 archive.
#[test]
fn mc_campaign_runs_an_arbitrary_blueprint() {
    if !local_data_present() {
        eprintln!("skip: no local data at {}", data_server::DEFAULT_DATA_PATH);
        return;
    }
    const FROM_MS: &str = "1735689600000";
    const TO_MS: &str = "1767225599000";
    let (cwd, _g) = fresh_project();
    let fixture = format!("{}/tests/fixtures/r_breakout_open.json", env!("CARGO_MANIFEST_DIR"));
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args([
            "mc", &fixture, "--real", "GER40",
            "--axis", "r_breakout_signal.channel_length=10,20",
            "--stop-length", "3", "--stop-k", "2.0",
            "--block-len", "5", "--resamples", "100", "--seed", "7",
            "--from", FROM_MS, "--to", TO_MS,
        ])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura");
    if out.status.code() == Some(1) {
        let stderr = String::from_utf8_lossy(&out.stderr);
        assert!(
            stderr.contains("no local data") || stderr.contains("no recorded geometry"),
            "exit 1 must be a data refusal, got: {stderr}"
        );
        eprintln!("skip: no local GER40 data");
        return;
    }
    assert_eq!(
        out.status.code(),
        Some(0),
        "exit: {:?} stderr: {}",
        out.status,
        String::from_utf8_lossy(&out.stderr)
    );
    let stdout = String::from_utf8(out.stdout).expect("utf-8 stdout");
    let grade_line = stdout
        .lines()
        .find(|l| l.starts_with("{\"mc_r_bootstrap\":"))
        .unwrap_or_else(|| panic!("the mc_r_bootstrap grade line: {stdout}"));
    let v: serde_json::Value = serde_json::from_str(grade_line).expect("grade line parses as JSON");
    assert_eq!(v["mc_r_bootstrap"]["n_resamples"].as_u64(), Some(100), "resamples flow: {grade_line}");
    assert!(v["mc_r_bootstrap"]["e_r"]["mean"].is_number(), "bootstrap E[R] present: {grade_line}");
}

/// #220 weld-gone proof for generalize: one mean-reversion candidate graded
/// across two instruments — two axes with names the CLI has never heard of —
/// one ganged I64 window and an independent F64 band. Gated on the GER40/USDJPY
/// Sept-2024 archive.
#[test]
fn generalize_campaign_runs_an_arbitrary_blueprint() {
    if !local_data_present() {
        eprintln!("skip: no local data at {}", data_server::DEFAULT_DATA_PATH);
        return;
    }
    let (cwd, _g) = fresh_project();
    let fixture = format!("{}/tests/fixtures/r_meanrev_open.json", env!("CARGO_MANIFEST_DIR"));
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args([
            "generalize", &fixture, "--real", "GER40,USDJPY",
            "--axis", "r_meanrev_signal.window=20",
            "--axis", "r_meanrev_signal.band.factor=2.0",
            "--stop-length", "3", "--stop-k", "2.0",
            "--from", GER40_SEPT2024_FROM_MS, "--to", GER40_SEPT2024_TO_MS,
        ])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura");
    if out.status.code() == Some(1) {
        let stderr = String::from_utf8_lossy(&out.stderr);
        assert!(
            stderr.contains("no local data") || stderr.contains("no recorded geometry"),
            "exit 1 must be a data refusal, got: {stderr}"
        );
        eprintln!("skip: no local GER40/USDJPY data");
        return;
    }
    assert_eq!(
        out.status.code(),
        Some(0),
        "exit: {:?} stderr: {}",
        out.status,
        String::from_utf8_lossy(&out.stderr)
    );
    let stdout = String::from_utf8(out.stdout).expect("utf-8 stdout");
    assert!(stdout.contains("\"generalize\":"), "aggregate object: {stdout}");
    assert!(stdout.contains("\"n_instruments\":2"), "two instruments graded: {stdout}");
    assert!(stdout.contains("\"worst_case\":"), "worst-case floor present: {stdout}");
}

/// #220: the walkforward campaign path requires >= 1 --axis (exit 2, data-free).
#[test]
fn walkforward_campaign_refuses_missing_axis() {
    let cwd = temp_cwd("walkforward-campaign-no-axis");
    let fixture = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args(["walkforward", &fixture, "--real", "GER40", "--stop-length", "14", "--stop-k", "2.0"])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura");
    assert_eq!(out.status.code(), Some(2), "missing --axis is a usage refusal");
    let stderr = String::from_utf8_lossy(&out.stderr);
    assert!(stderr.contains("requires >= 1 --axis"), "axis-arity refusal: {stderr}");
}

/// #220: mc's two modes are disjoint — --seeds (synthetic seed family) with
/// --real (campaign pipeline) is a cross-mode usage error (exit 2, data-free).
#[test]
fn mc_campaign_refuses_seeds_with_real() {
    let cwd = temp_cwd("mc-campaign-seeds-mix");
    let fixture = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args([
            "mc", &fixture, "--real", "GER40", "--seeds", "3",
            "--axis", "sma_signal.fast.length=3", "--axis", "sma_signal.slow.length=12",
            "--stop-length", "14", "--stop-k", "2.0",
        ])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura");
    assert_eq!(out.status.code(), Some(2), "--seeds with --real is a cross-mode usage refusal");
    let stderr = String::from_utf8_lossy(&out.stderr);
    assert!(stderr.contains("Usage: aura mc"), "usage refusal: {stderr}");
}

/// #220: a raw-form axis name is refused before any data access on the migrated
/// verbs, exactly as sweep's shipped refusal — echoing what the user typed, with
/// no store litter (mirrors aura_sweep_real_blueprint_refuses_a_raw_form_axis_name_before_data_access).
#[test]
fn walkforward_campaign_refuses_a_raw_form_axis_name() {
    let cwd = temp_cwd("walkforward-campaign-raw-axis");
    let fixture = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args([
            "walkforward", &fixture, "--real", "GER40",
            "--axis", "fast.length=3,5", "--axis", "sma_signal.slow.length=12,20",
            "--stop-length", "14", "--stop-k", "2.0",
        ])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura");
    assert_eq!(out.status.code(), Some(2), "a raw-form axis name is refused before any data access");
    let stderr = String::from_utf8_lossy(&out.stderr);
    assert!(stderr.contains("axis \"fast.length\""), "echoes the raw name verbatim: {stderr}");
    assert!(!cwd.join("runs").exists(), "no store litter on the refusal path");
}

/// #220: an unknown axis name is refused at the dispatch boundary on the
/// migrated verbs (mirrors aura_sweep_rejects_an_unknown_axis).
#[test]
fn mc_campaign_refuses_an_unknown_axis() {
    let cwd = temp_cwd("mc-campaign-unknown-axis");
    let fixture = format!("{}/examples/r_breakout.json", env!("CARGO_MANIFEST_DIR"));
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args([
            "mc", &fixture, "--real", "GER40",
            "--axis", "r_breakout_signal.nope.length=1",
            "--stop-length", "14", "--stop-k", "2.0",
        ])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura");
    assert_eq!(out.status.code(), Some(2), "an unknown axis is refused at the dispatch boundary");
    let stderr = String::from_utf8_lossy(&out.stderr);
    assert!(
        stderr.contains("axis \"r_breakout_signal.nope.length\""),
        "names the offending axis: {stderr}"
    );
}

/// #220: the mc campaign path also requires >= 1 --axis before any archive
/// access (exit 2, data-free) — the same arity gate walkforward enforces
/// (`walkforward_campaign_refuses_missing_axis`) and generalize enforces
/// (`generalize_refuses_no_axis`), but which Task 5 left untested for mc: the
/// `axes.is_empty()` check lives in `dispatch_mc`'s `--real` arm alone, so a
/// regression dropping it would otherwise surface only once real data hits the
/// R-bootstrap pipeline, not at argv-parse time.
#[test]
fn mc_campaign_refuses_missing_axis() {
    let cwd = temp_cwd("mc-campaign-no-axis");
    let fixture = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args(["mc", &fixture, "--real", "GER40", "--stop-length", "14", "--stop-k", "2.0"])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura");
    assert_eq!(out.status.code(), Some(2), "missing --axis is a usage refusal");
    let stderr = String::from_utf8_lossy(&out.stderr);
    assert!(stderr.contains("Usage: aura mc"), "usage refusal: {stderr}");
    assert!(!cwd.join("runs").exists(), "no store litter on the refusal path");
}

/// `aura run` on the shipped closed example reproduces the built-in r-sma grade
/// (#159): the example is the proven data successor to `--harness r-sma`. Survives
/// the Cut-1b builder deletion (depends only on the example + generic run path).
#[test]
fn shipped_r_sma_example_reproduces_the_builtin_grade() {
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args(["run", "examples/r_sma.json"])
        .output()
        .expect("spawn aura run example");
    assert_eq!(
        out.status.code(),
        Some(0),
        "exit: {:?} stderr={}",
        out.status,
        String::from_utf8_lossy(&out.stderr)
    );
    let stdout = String::from_utf8(out.stdout).expect("utf-8");
    // the same metrics r_sma_single_run_output_golden pins for --harness r-sma
    assert!(stdout.contains("\"expectancy_r\":1.2710005136982836"), "{stdout}");
    assert!(stdout.contains("\"sqn\":3.141496526818299"), "{stdout}");
    assert!(stdout.contains("\"total_pips\":0.34185000000002036"), "{stdout}");
    assert!(stdout.contains("\"n_trades\":3"), "{stdout}");
}

/// Characterization pin (#252, byte-identity anchor before `probe_window`'s
/// internals swap from a full-window drain to `aura_ingest::archive_extent`'s
/// boundary-month derivation). Captured against the PRE-#252 drain-based code
/// over `fresh_project_with_data`'s hermetic two-symbol synthetic archive
/// (SYMA spans 2024-01..08, SYMB 2024-03..06, weekday 08:00-16:00 UTC bars):
/// the resolved `campaign_window_ms(full_window)` span persists verbatim into
/// the generated campaign document's `data.windows[0]`, exactly like the
/// sibling `generalize_without_explicit_window_resolves_the_intersection_of_all_symbols`
/// probe. Three cases: SYMA's and SYMB's full (no-window) spans, and an
/// explicit `--from`/`--to` sub-window straddling three month files
/// (2024-03-15..2024-05-15, so `filter_files` alone would load 3 files under
/// the old drain). The first bar of each captured span carries a known -1ms
/// float round-trip artifact (the synthetic fixture's Delphi-`TDateTime`
/// pack/unpack in `data-server`'s own record format, not a `probe_window` bug)
/// — this test does not explain it, only pins it exactly, since the
/// post-swap code must reproduce the identical bytes (same underlying decoder,
/// only fewer files touched to reach them).
#[test]
fn probe_window_resolves_the_pinned_span_before_the_252_swap() {
    let (cwd, _g) = fresh_project_with_data();
    let fixture = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let window_of = |name: &str| -> (i64, i64) {
        let dir = cwd.join("runs").join("campaigns");
        let path = std::fs::read_dir(&dir)
            .expect("campaigns dir exists after a run")
            .filter_map(|e| e.ok())
            .map(|e| e.path())
            .find(|p| {
                let doc: serde_json::Value =
                    serde_json::from_str(&std::fs::read_to_string(p).expect("read campaign doc"))
                        .expect("campaign doc parses as JSON");
                doc["name"].as_str() == Some(name)
            })
            .unwrap_or_else(|| panic!("no campaign document named {name:?} in {}", dir.display()));
        let doc: serde_json::Value =
            serde_json::from_str(&std::fs::read_to_string(&path).expect("read campaign doc"))
                .expect("campaign doc parses as JSON");
        let w = &doc["data"]["windows"][0];
        (
            w["from_ms"].as_i64().expect("from_ms is an integer"),
            w["to_ms"].as_i64().expect("to_ms is an integer"),
        )
    };
    let sweep = |symbol: &str, name: &str, from: Option<&str>, to: Option<&str>| {
        let mut args = vec![
            "sweep", fixture.as_str(), "--real", symbol,
            "--axis", "sma_signal.fast.length=3", "--axis", "sma_signal.slow.length=12",
            "--name", name,
        ];
        if let Some(f) = from {
            args.push("--from");
            args.push(f);
        }
        if let Some(t) = to {
            args.push("--to");
            args.push(t);
        }
        let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
            .args(&args)
            .current_dir(&cwd)
            .output()
            .expect("spawn aura sweep");
        assert_eq!(
            out.status.code(), Some(0),
            "window-probe sweep {name:?} must run to exit 0: {}",
            String::from_utf8_lossy(&out.stderr)
        );
    };

    sweep("SYMA", "probe-252-no-window-syma", None, None);
    assert_eq!(
        window_of("probe-252-no-window-syma"),
        (1_704_095_999_999, 1_725_033_540_000),
        "SYMA's full (no-window) resolved span"
    );

    sweep("SYMB", "probe-252-no-window-symb", None, None);
    assert_eq!(
        window_of("probe-252-no-window-symb"),
        (1_709_279_999_999, 1_719_590_340_000),
        "SYMB's full (no-window) resolved span"
    );

    // 2024-03-15 00:00 UTC .. 2024-05-15 00:00 UTC: straddles the March, April,
    // and May files (three boundary months under the old file-range drain).
    sweep("SYMA", "probe-252-straddle", Some("1710460800000"), Some("1715731200000"));
    assert_eq!(
        window_of("probe-252-straddle"),
        (1_710_489_599_999, 1_715_702_340_000),
        "explicit sub-window straddling month boundaries, clipped to the archive's actual bars"
    );
}

/// Characterization pin (#252, hostless zero-bar case): a window that falls
/// entirely on a weekend inside a COVERED month (2024-03-02/03, both inside
/// SYMA's Jan-Aug 2024 span, weekday-only bars) still refuses with the SAME
/// exit-1 "no data ... in the requested window" message `probe_window`'s
/// drain-based emptiness check raises today — `archive_extent` must reach the
/// identical conclusion by finding zero matching months in range, not by
/// draining a stream, and the refusal wording is part of what must not move
/// (mirrors `run_real_empty_window_refusal_names_the_window_not_the_symbol`,
/// hostless here since the property (an empty resolved sub-window still
/// refuses) needs no full archive, only a covered-but-empty range).
#[test]
fn probe_window_still_refuses_a_hostless_zero_bar_window() {
    let (cwd, _g) = fresh_project_with_data();
    let fixture = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args([
            "sweep", &fixture, "--real", "SYMA",
            "--axis", "sma_signal.fast.length=3", "--axis", "sma_signal.slow.length=12",
            "--name", "probe-252-zero-bars",
            "--from", "1709337600000", // 2024-03-02 00:00 UTC (Saturday)
            "--to", "1709424000000",   // 2024-03-03 00:00 UTC (Sunday)
        ])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura sweep");
    assert_eq!(
        out.status.code(), Some(1),
        "a zero-bar window inside covered archive months still refuses with exit 1: stderr={}",
        String::from_utf8_lossy(&out.stderr)
    );
    let stderr = String::from_utf8_lossy(&out.stderr);
    assert!(
        stderr.contains("no data") && stderr.contains("window"),
        "must name the empty window, not a missing symbol: {stderr}"
    );
    assert!(
        !cwd.join("runs").join("campaigns").exists()
            || std::fs::read_dir(cwd.join("runs").join("campaigns")).unwrap().count() == 0,
        "a refused zero-bar window must not register a generated campaign document"
    );
}

/// `aura run` on the shipped closed r-meanrev example (#159 cut 3) actually loads
/// and grades through the real, separately-linked `aura` binary — not merely
/// in-process, as `r_meanrev_example_loaded_runs_identically_to_the_carved_signal`
/// (main.rs) and this test's own grade-equivalence check both do by calling
/// `run_signal_r`/`blueprint_from_json` inside the *test* binary. This is the
/// CLI-seam companion `shipped_r_sma_example_reproduces_the_builtin_grade` has for
/// r-sma: it pins that the `Scale`-based band, freshly added to the closed
/// vocabulary this iteration, resolves and runs cleanly through the production
/// vocabulary lookup + arg-parsing path a real invocation takes, not just through
/// the test harness's direct function calls. The synthetic stream never crosses
/// the band at this window/k, so the grade is genuinely all-zero (no trade) —
/// still a deterministic, meaningful pin: a regression that broke `Scale`'s CLI
/// wiring (e.g. a roster/registration mismatch) would surface as a non-zero exit
/// or a load error here, not as a metrics drift.
#[test]
fn shipped_r_meanrev_example_runs_end_to_end_via_aura_run() {
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args(["run", "examples/r_meanrev.json"])
        .output()
        .expect("spawn aura run example");
    assert_eq!(
        out.status.code(),
        Some(0),
        "exit: {:?} stderr={}",
        out.status,
        String::from_utf8_lossy(&out.stderr)
    );
    let stdout = String::from_utf8(out.stdout).expect("utf-8");
    assert!(stdout.contains("\"topology_hash\":\""), "manifest carries the topology hash: {stdout}");
    assert!(stdout.contains("\"expectancy_r\":0.0"), "{stdout}");
    assert!(stdout.contains("\"n_trades\":0"), "{stdout}");
    assert!(stdout.contains("\"total_pips\":0.0"), "{stdout}");
}

/// Property (#264, coverage cut 1): `aura data coverage <SYMBOL>` surfaces the
/// archive's month COVERAGE from the monthly file index — the first and last
/// present month AND every interior missing-month range (a gap) — so the Copper
/// failure mode is visible up front. Copper had files from 2016-08 but was
/// missing 2018-01..2018-10, so a first/last-bounds check passed while a
/// window opening in the hole had no data, aborting a multi-hour campaign
/// mid-run; a per-symbol gap report would have caught it before the run. Over a
/// hermetic synthetic archive whose `GAPSYM` has 2024-01..02 and 2024-05..06
/// present and 2024-03..04 absent, the verb exits 0, frames the span by its
/// first (2024-01) and last (2024-06) present month, and reports the interior
/// hole as a single collapsed month-range line `missing: 2024-03..2024-04`
/// (one line for the whole contiguous gap, as Copper's ten-month hole would be
/// one line — not one line per missing month).
#[test]
fn data_coverage_reports_month_span_and_interior_gap_range() {
    let (cwd, _g) = fresh_project_with_gapped_data();
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args(["data", "coverage", "GAPSYM"])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura data coverage");
    assert_eq!(
        out.status.code(),
        Some(0),
        "exit: {:?}, stderr: {}",
        out.status,
        String::from_utf8_lossy(&out.stderr)
    );
    let stdout = String::from_utf8(out.stdout).expect("utf-8 stdout");
    // The interior gap is the headline — the Copper failure the verb exists to
    // surface — reported as one collapsed range for the whole contiguous hole.
    assert!(
        stdout.contains("missing: 2024-03..2024-04"),
        "interior gap month-range reported as one line: {stdout}"
    );
    // The span bounds frame the gap: first and last present month.
    assert!(
        stdout.contains("2024-01") && stdout.contains("2024-06"),
        "first (2024-01) and last (2024-06) present month reported: {stdout}"
    );
}

/// Property (#264, list cut 2): `aura data list` enumerates the symbols the
/// archive contains, one per line, sorted — the discovery step a campaign
/// author (or an agent operating through the CLI alone, the deployment
/// posture) needs before `aura data coverage <symbol>`, since without it the
/// inventory is obtainable only by listing the archive directory outside
/// aura. The archive root resolves through the same project path as `coverage`
/// (`Env::data_path` — a project-local `[paths] data` override when set). Over
/// a hermetic synthetic archive holding two symbols (`SYMA` 2024-01..08 and
/// `SYMB` 2024-03..06), the verb exits 0 and prints EXACTLY the two symbol
/// names in ascending order, one per line — `SYMA` before `SYMB`, nothing
/// else.
#[test]
fn data_list_enumerates_archive_symbols_one_per_line_sorted() {
    let (cwd, _g) = fresh_project_with_data();
    let out = std::process::Command::new(env!("CARGO_BIN_EXE_aura"))
        .args(["data", "list"])
        .current_dir(&cwd)
        .output()
        .expect("spawn aura data list");
    assert_eq!(
        out.status.code(),
        Some(0),
        "exit: {:?}, stderr: {}",
        out.status,
        String::from_utf8_lossy(&out.stderr)
    );
    let stdout = String::from_utf8(out.stdout).expect("utf-8 stdout");
    // The whole contract: the two archive symbols, sorted, one per line, and
    // nothing else — this is the inventory a campaign author reads before
    // scoping an instrument matrix.
    assert_eq!(stdout, "SYMA\nSYMB\n", "sorted symbol enumeration: {stdout:?}");
}
