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

//! Integration test: a family-emitting subcommand must survive its stdout
//! reader closing early (the most natural CLI piping — `| head`, `| less`, a
//! closed UI pane). Drives the built `aura` binary and closes the read end of
//! its stdout pipe before the writer finishes, then asserts the EPIPE is
//! handled cleanly, never a panic.

use std::io::Read;
use std::path::Path;
use std::process::{Command, Stdio};

mod common;
use common::{fresh_project_with_data, ScratchGuard, ScratchPath};

/// 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");

fn run_code_in(dir: &Path, args: &[&str]) -> (String, Option<i32>) {
    let out = Command::new(BIN).args(args).current_dir(dir).output().expect("binary runs");
    let text = format!(
        "{}{}",
        String::from_utf8_lossy(&out.stdout),
        String::from_utf8_lossy(&out.stderr)
    );
    (text, out.status.code())
}

fn write_doc(dir: &Path, name: &str, text: &str) -> std::path::PathBuf {
    let p = dir.join(name);
    std::fs::write(&p, text).expect("write doc");
    p
}

/// The minimal executable pipeline (one sweep stage) — copied verbatim from
/// `research_docs.rs`'s constant of the same name.
const SWEEP_ONLY_PROCESS_DOC: &str = r#"{
  "format_version": 1,
  "kind": "process",
  "name": "sweep-only",
  "pipeline": [ { "block": "std::sweep", "metric": "sqn_normalized", "select": "argmax" } ]
}"#;

/// Seed one blueprint into the built demo project's store via `aura graph
/// register` and return its content id — copied verbatim from
/// `research_docs.rs`'s recipe of the same name (#319 — the retired `sweep`
/// side-effect seeding this used to ride played no role in the returned
/// content id, a pure function of the loaded blueprint's canonical bytes;
/// `name` labels the registered id, mirroring the old per-call sweep family
/// name).
fn seed_blueprint(dir: &Path, name: &str) -> String {
    let closed_bp = format!("{}/examples/r_sma.json", env!("CARGO_MANIFEST_DIR"));
    let (out, code) = run_code_in(dir, &["graph", "register", &closed_bp, "--name", name]);
    assert_eq!(code, Some(0), "seed register failed: {out}");
    out.lines()
        .find(|l| l.starts_with("registered blueprint "))
        .expect("register line")
        .trim_start_matches("registered blueprint ")
        .split(' ')
        .next()
        .expect("id")
        .trim_start_matches("content:")
        .to_string()
}

/// Register `doc` as a process document in the project store; returns its
/// id — copied verbatim from `research_docs.rs`'s recipe of the same name.
fn register_process_doc(dir: &Path, file: &str, doc: &str) -> String {
    write_doc(dir, file, doc);
    let (out, code) = run_code_in(dir, &["process", "register", file]);
    assert_eq!(code, Some(0), "process register failed: {out}");
    out.lines()
        .find(|l| l.starts_with("registered process "))
        .expect("register line")
        .trim_start_matches("registered process ")
        .split(' ')
        .next()
        .expect("id")
        .trim_start_matches("content:")
        .to_string()
}

/// A one-strategy, two-axis (2×2) campaign document over `bp_id`/`proc_id` —
/// copied verbatim from `exec.rs`'s `campaign_doc_json_for` of the same
/// shape, which yields exactly four family members (the four-line stream
/// this test's property needs).
fn campaign_doc_json_for(instrument: &str, bp_id: &str, proc_id: &str, window: (i64, i64)) -> String {
    format!(
        r#"{{
  "format_version": 1,
  "kind": "campaign",
  "name": "broken-pipe",
  "data": {{ "instruments": ["{instrument}"], "windows": [ {{ "from_ms": {from}, "to_ms": {to} }} ] }},
  "strategies": [ {{ "ref": {{ "content_id": "{bp_id}" }},
                    "axes": {{ "fast.length": {{ "kind": "I64", "values": [2, 4] }},
                              "slow.length": {{ "kind": "I64", "values": [8, 16] }} }} }} ],
  "process": {{ "ref": {{ "content_id": "{proc_id}" }} }},
  "seed": 7,
  "presentation": {{ "persist_taps": [], "emit": ["family_table"] }}
}}"#,
        from = window.0,
        to = window.1,
    )
}

/// Property: a family-emitting command whose stdout reader closes early exits on
/// the EPIPE cleanly — it does NOT panic (`failed printing to stdout: Broken
/// pipe`) and does NOT exit 101. `aura exec <campaign.json>` streams four
/// family-member JSON lines via `println!` (a 2×2 axis grid — #319: the
/// surviving surface for what a bare `aura sweep` demo grid used to stream);
/// when the reader closes stdout after the first line, the next write hits a
/// closed pipe. Rust's runtime sets SIGPIPE to SIG_IGN at startup, so that
/// write returns EPIPE and `println!` panics rather than the process
/// terminating quietly — exactly the behaviour a CLI must not have. This is
/// the contract: any aura subcommand can be piped into `| head` / `| less` /
/// a UI pane that goes away, and it must finish quietly, not panic.
#[test]
fn family_emitting_command_survives_early_reader_close() {
    let (dir, _fixture) = fresh_project_with_data();
    let runs_dir = dir.join("runs");
    std::fs::remove_dir_all(&runs_dir).ok();
    let _cleanup = ScratchGuard(vec![
        ScratchPath::Dir(runs_dir),
        ScratchPath::File(dir.join("broken-pipe.process.json")),
        ScratchPath::File(dir.join("broken-pipe.campaign.json")),
    ]);
    let bp_id = seed_blueprint(&dir, "broken-pipe-seed");
    let proc_id = register_process_doc(&dir, "broken-pipe.process.json", SWEEP_ONLY_PROCESS_DOC);
    let doc = campaign_doc_json_for("SYMA", &bp_id, &proc_id, (1709251200000, 1719791999999));
    write_doc(&dir, "broken-pipe.campaign.json", &doc);

    let mut child = Command::new(BIN)
        .args(["exec", "broken-pipe.campaign.json"])
        .current_dir(&dir)
        .stdout(Stdio::piped())
        .stderr(Stdio::piped())
        .spawn()
        .expect("spawn aura exec");

    // Read just enough to guarantee the writer is mid-stream, then drop the read
    // handle. Dropping it closes the read end of the pipe; the next `println!`
    // in the writer then writes to a closed pipe and gets EPIPE.
    {
        let mut stdout = child.stdout.take().expect("child stdout piped");
        let mut buf = [0u8; 64];
        // One short read advances the stream; we deliberately ignore the rest.
        let _ = stdout.read(&mut buf);
        // `stdout` drops here, closing the read end of the pipe.
    }

    // Drain stderr so the child cannot block writing the panic message, then reap.
    let mut stderr = String::new();
    if let Some(mut err) = child.stderr.take() {
        let _ = err.read_to_string(&mut stderr);
    }
    let status = child.wait().expect("reap aura exec");

    // The defect: stderr carries the broken-pipe panic.
    assert!(
        !stderr.contains("panicked"),
        "a closed stdout reader must not panic the writer; stderr was: {stderr:?}"
    );
    assert!(
        !stderr.contains("Broken pipe"),
        "the broken pipe must be handled cleanly, not surfaced as an error; stderr was: {stderr:?}"
    );

    // And the exit is the clean broken-pipe outcome, never the panic's 101.
    // A clean SIGPIPE termination is signal 13 (code is None); a deliberate
    // clean-exit code is anything but 101. The only forbidden outcome is the
    // panic abort.
    assert_ne!(
        status.code(),
        Some(101),
        "a closed stdout reader must not exit with the panic code 101; status: {status:?}"
    );
}