//! End-to-end observation tests: run the real nash binary against a fake
//! Nucleic control-socket sink and assert on the shell-batch events it posts
//! (docs/NASH.md §6.1, §10).

use std::io::{Read, Write};
use std::os::unix::net::UnixListener;
use std::process::Command;
use std::sync::atomic::{AtomicU32, Ordering};
use std::sync::mpsc;
use std::time::Duration;

/// Conservative floor for `sockaddr_un.sun_path` (104 on macOS, 108 on Linux),
/// leaving room for the NUL terminator. A socket path over this fails to bind
/// with "path must be shorter than SUN_LEN".
const MAX_SOCKET_PATH: usize = 100;

/// A fresh, uniquely named scratch directory for one test.
///
/// `$TMPDIR` is a long `/var/folders/…` path on macOS and can be set anywhere
/// by the caller, so a socket placed under it can blow the `sun_path` limit —
/// keeping the directory name short is necessary but not sufficient. Fall back
/// to `/tmp` whenever the resulting socket path would not fit; the bind limit
/// applies to the path we pass, and `/tmp` is short on every platform we run on.
///
/// Names combine the pid with a per-process counter, so concurrent test threads
/// never collide; any leftover directory from an earlier run with a recycled
/// pid is removed rather than inherited (a stale socket file would otherwise
/// fail the bind with `AlreadyExists`).
fn scratch_dir(prefix: &str) -> std::path::PathBuf {
    static COUNTER: AtomicU32 = AtomicU32::new(0);
    let name = format!(
        "{prefix}-{}-{}",
        std::process::id(),
        COUNTER.fetch_add(1, Ordering::Relaxed)
    );

    let dir = [std::env::temp_dir(), std::path::PathBuf::from("/tmp")]
        .into_iter()
        .map(|base| base.join(&name))
        .find(|dir| dir.join("control.sock").as_os_str().len() <= MAX_SOCKET_PATH)
        .expect("no temp base short enough for a unix socket path");

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

struct Sink {
    dir: std::path::PathBuf,
    rx: mpsc::Receiver<(Option<String>, serde_json::Value)>,
    /// Connections accepted so far — how the keep-alive test tells one reused connection from
    /// one per batch (docs/NASH_STREAM_PERF_PLAN.md §P4.1).
    connections: std::sync::Arc<AtomicU32>,
}

impl Sink {
    /// Starts a unix-socket HTTP sink; each posted batch is parsed and sent
    /// through the channel along with its Authorization header.
    fn start() -> Self {
        let dir = scratch_dir("nash-test");
        let socket_path = dir.join("control.sock");
        let listener = UnixListener::bind(&socket_path)
            .unwrap_or_else(|e| panic!("bind {} ({} bytes): {e}", socket_path.display(), socket_path.as_os_str().len()));
        let (tx, rx) = mpsc::channel();
        let connections = std::sync::Arc::new(AtomicU32::new(0));
        let accepted = std::sync::Arc::clone(&connections);
        std::thread::spawn(move || {
            for stream in listener.incoming() {
                let Ok(mut stream) = stream else { break };
                accepted.fetch_add(1, Ordering::Relaxed);
                let tx = tx.clone();
                std::thread::spawn(move || {
                    let mut buf = Vec::new();
                    let mut chunk = [0u8; 4096];
                    // Read until headers + declared body length are complete, answer, and keep
                    // reading: nash reuses one connection for every batch of a shell's life
                    // (docs/NASH_STREAM_PERF_PLAN.md §P4.1), exactly as the host's route does.
                    loop {
                        if let Some(request) = request_len(&buf) {
                            let auth = header(&buf, "authorization");
                            if let Some(body) = full_body(&buf) {
                                if let Ok(json) = serde_json::from_slice(body) {
                                    let _ = tx.send((auth, json));
                                }
                            }
                            if stream
                                .write_all(
                                    b"HTTP/1.1 202 Accepted\r\nContent-Length: 0\r\nConnection: keep-alive\r\n\r\n",
                                )
                                .is_err()
                            {
                                break;
                            }
                            buf.drain(..request);
                            continue;
                        }
                        match stream.read(&mut chunk) {
                            Ok(0) => break,
                            Ok(n) => buf.extend_from_slice(&chunk[..n]),
                            Err(_) => break,
                        }
                    }
                });
            }
        });
        Self {
            dir,
            rx,
            connections,
        }
    }

    fn socket(&self) -> std::path::PathBuf {
        self.dir.join("control.sock")
    }

    /// Collects events from batches until `pred` is satisfied or ~5s elapse.
    fn events_until(
        &self,
        pred: impl Fn(&[serde_json::Value]) -> bool,
    ) -> Vec<serde_json::Value> {
        let mut events = Vec::new();
        let deadline = std::time::Instant::now() + Duration::from_secs(5);
        while std::time::Instant::now() < deadline {
            match self.rx.recv_timeout(Duration::from_millis(250)) {
                Ok((auth, batch)) => {
                    assert_eq!(
                        auth.as_deref(),
                        Some("Bearer test-token"),
                        "batch must carry the bearer token"
                    );
                    assert_eq!(batch["type"], "shell-batch");
                    assert_eq!(batch["source"], "nash");
                    assert_eq!(batch["sessionId"], "sess-1");
                    assert!(batch["shellId"].is_string());
                    events.extend(batch["events"].as_array().cloned().unwrap_or_default());
                    if pred(&events) {
                        break;
                    }
                }
                Err(_) => {
                    if pred(&events) {
                        break;
                    }
                }
            }
        }
        events
    }

    fn next_batch(&self) -> serde_json::Value {
        let (auth, batch) = self
            .rx
            .recv_timeout(Duration::from_secs(5))
            .expect("shell batch");
        assert_eq!(auth.as_deref(), Some("Bearer test-token"));
        batch
    }
}

impl Drop for Sink {
    /// Don't leave the socket behind: a stale one under a recycled pid is what
    /// turned a rerun into an `AlreadyExists` bind failure.
    fn drop(&mut self) {
        let _ = std::fs::remove_dir_all(&self.dir);
    }
}

/// Total bytes of the first complete request in `buf` (head + body), or None while it is still
/// arriving — what lets one connection carry the next batch after this one is answered.
fn request_len(buf: &[u8]) -> Option<usize> {
    let headers_end = buf.windows(4).position(|w| w == b"\r\n\r\n")? + 4;
    let len: usize = header(buf, "content-length")?.parse().ok()?;
    (buf.len() >= headers_end + len).then_some(headers_end + len)
}

fn full_body(buf: &[u8]) -> Option<&[u8]> {
    let headers_end = buf.windows(4).position(|w| w == b"\r\n\r\n")? + 4;
    let len: usize = header(buf, "content-length")?.parse().ok()?;
    let body = &buf[headers_end..];
    (body.len() >= len).then(|| &body[..len])
}

fn header(buf: &[u8], name: &str) -> Option<String> {
    let text = String::from_utf8_lossy(buf);
    text.lines()
        .take_while(|l| !l.is_empty())
        .find_map(|l| {
            let (k, v) = l.split_once(':')?;
            k.trim().eq_ignore_ascii_case(name).then(|| v.trim().to_string())
        })
}

fn nash_command(sink: &Sink, script: &str) -> Command {
    let mut command = Command::new(env!("CARGO_BIN_EXE_nash"));
    command
        .args(["-c", script])
        .env("NUCLEIC_SHELL_SOCKET", sink.socket())
        .env("NUCLEIC_HOOK_TOKEN", "test-token")
        .env("NUCLEIC_SESSION_ID", "sess-1")
        .env("NUCLEIC_SHELL_ENVIRONMENT_KIND", "agent")
        .env("NUCLEIC_SHELL_ENVIRONMENT_ID", "agent")
        .env("NUCLEIC_SHELL_ENVIRONMENT_LABEL", "Agent")
        .env_remove("NUCLEIC_SHELL_PARENT");
    command
}

fn run_nash(sink: &Sink, script: &str) -> std::process::ExitStatus {
    nash_command(sink, script).status().unwrap()
}

fn find<'a>(
    events: &'a [serde_json::Value],
    kind: &str,
    pred: impl Fn(&serde_json::Value) -> bool,
) -> Option<&'a serde_json::Value> {
    events
        .iter()
        .find(|e| e["kind"] == kind && pred(e))
}

#[test]
fn exec_events_flow() {
    let sink = Sink::start();
    let status = run_nash(&sink, "echo observed; false");
    assert_eq!(status.code(), Some(1));

    let events = sink.events_until(|evs| {
        find(evs, "exec", |e| e["argv"][0] == "echo").is_some()
            && find(evs, "exec", |e| e["argv"][0] == "false").is_some()
    });

    let echo = find(&events, "exec", |e| e["argv"][0] == "echo").expect("echo exec event");
    assert_eq!(echo["argv"][1], "observed");
    assert_eq!(echo["exitCode"], 0);
    assert!(echo["cwd"].is_string());
    assert!(echo["durationMs"].is_u64());

    let false_ev = find(&events, "exec", |e| e["argv"][0] == "false").expect("false exec event");
    assert_eq!(false_ev["exitCode"], 1);

    // A command that returns promptly is reported once, on exit. Announcing every command as
    // "running" would double the exec traffic to say so about commands already long over.
    assert!(
        find(&events, "exec-start", |_| true).is_none(),
        "fast commands must not be announced as running"
    );
}

/// A command that has not returned is reported while it runs (docs/NASH.md §5.1) — otherwise the
/// host hears nothing at all during exactly the commands worth watching, and a five-minute build
/// first appears once it is over.
#[test]
fn long_running_command_announced_before_it_returns() {
    let sink = Sink::start();
    // Spawned, not run to completion: the point of the event is that it arrives *mid-command*.
    let mut child = nash_command(&sink, "sleep 3").spawn().unwrap();

    let events = sink.events_until(|evs| find(evs, "exec-start", |e| e["argv"][0] == "sleep").is_some());
    let start = find(&events, "exec-start", |e| e["argv"][0] == "sleep")
        .expect("exec-start while the command is still running");
    assert!(
        child.try_wait().unwrap().is_none(),
        "the announcement arrived only after the command had already returned"
    );
    assert_eq!(start["argv"][1], "3");
    assert!(start["cwd"].is_string());
    // How long it had *already* been running, so a host counts elapsed time from the command's
    // own start rather than from the moment the batch reached it.
    let elapsed = start["elapsedMs"].as_u64().expect("elapsedMs on exec-start");
    assert!((400..3_000).contains(&elapsed), "elapsedMs was {elapsed}");
    // Nothing about how it ended — that is what the exec event is for.
    assert!(start.get("exitCode").is_none());
    assert!(start.get("durationMs").is_none());
    let token = start["token"].as_u64().expect("gate token on exec-start");

    child.wait().unwrap();
    let rest = sink.events_until(|evs| find(evs, "exec", |e| e["argv"][0] == "sleep").is_some());
    let done = find(&rest, "exec", |e| e["argv"][0] == "sleep").expect("exec event on return");
    // The join a host folds the two into one row on: same command run, same token.
    assert_eq!(done["token"].as_u64(), Some(token));
    assert_eq!(done["exitCode"], 0);
    // Announced once, however many sweeps the command outlived — a row that re-announced itself
    // every quarter second would be a new row every quarter second.
    let announcements = events.iter().chain(rest.iter()).filter(|e| {
        e["kind"] == "exec-start" && e["argv"][0] == "sleep"
    });
    assert_eq!(announcements.count(), 1);
}

#[test]
fn emitter_environment_flows_on_batch_envelope() {
    let sink = Sink::start();
    let status = run_nash(&sink, "true");
    assert_eq!(status.code(), Some(0));

    let batch = sink.next_batch();
    assert_eq!(batch["environmentKind"], "agent");
    assert_eq!(batch["environmentId"], "agent");
    assert_eq!(batch["environmentLabel"], "Agent");
}

#[test]
fn external_command_exit_codes() {
    let sink = Sink::start();
    let status = run_nash(&sink, "/bin/echo ext; /usr/bin/env true; sh -c 'exit 7'");
    assert_eq!(status.code(), Some(7));

    let events = sink.events_until(|evs| {
        find(evs, "exec", |e| e["argv"][0] == "sh").is_some()
    });
    let ext = find(&events, "exec", |e| e["argv"][0] == "/bin/echo").expect("external exec event");
    assert_eq!(ext["exitCode"], 0);
    let sh = find(&events, "exec", |e| e["argv"][0] == "sh").expect("sh exec event");
    assert_eq!(sh["exitCode"], 7);
}

#[test]
fn cd_and_export_events() {
    let sink = Sink::start();
    let status = run_nash(&sink, "cd /tmp && export NASH_E2E=1 OTHER=2 && unset OTHER");
    assert_eq!(status.code(), Some(0));

    let events = sink.events_until(|evs| {
        find(evs, "cd", |_| true).is_some()
            && find(evs, "export", |e| e["op"] == "unset").is_some()
    });

    let cd = find(&events, "cd", |_| true).expect("cd event");
    assert_eq!(cd["to"], "/tmp");
    assert!(cd["from"].is_string());

    let export = find(&events, "export", |e| e["op"] == "export").expect("export event");
    let names: Vec<_> = export["names"].as_array().unwrap().iter().collect();
    assert_eq!(names.len(), 2);
    assert_eq!(export["names"][0], "NASH_E2E");
    assert_eq!(export["names"][1], "OTHER");
    // Names only — values must never appear anywhere in the event.
    assert!(!serde_json::to_string(export).unwrap().contains('='));

    let unset = find(&events, "export", |e| e["op"] == "unset").expect("unset event");
    assert_eq!(unset["names"][0], "OTHER");
}

#[test]
fn pipeline_members_each_reported() {
    let sink = Sink::start();
    let status = run_nash(&sink, "printf 'a\\nb\\n' | wc -l | cat > /dev/null");
    assert_eq!(status.code(), Some(0));

    let events = sink.events_until(|evs| {
        ["printf", "wc", "cat"]
            .iter()
            .all(|c| find(evs, "exec", |e| &e["argv"][0] == c).is_some())
    });
    for cmd in ["printf", "wc", "cat"] {
        let ev = find(&events, "exec", |e| &e["argv"][0] == cmd)
            .unwrap_or_else(|| panic!("{cmd} exec event missing"));
        assert_eq!(ev["exitCode"], 0, "{cmd} should exit 0");
    }
}

#[test]
fn command_not_found_reported() {
    let sink = Sink::start();
    let status = run_nash(&sink, "definitely-not-a-command-xyz 2>/dev/null; true");
    assert_eq!(status.code(), Some(0));

    let events = sink.events_until(|evs| {
        find(evs, "exec", |e| e["argv"][0] == "definitely-not-a-command-xyz").is_some()
    });
    let ev = find(&events, "exec", |e| e["argv"][0] == "definitely-not-a-command-xyz")
        .expect("not-found exec event");
    assert_eq!(ev["exitCode"], 127);
}

#[test]
fn parse_failure_falls_back_to_bash() {
    let sink = Sink::start();
    let status = Command::new(env!("CARGO_BIN_EXE_nash"))
        .args(["-c", "if then fi"])
        .env("NUCLEIC_SHELL_SOCKET", sink.socket())
        .env("NUCLEIC_HOOK_TOKEN", "test-token")
        .env("NUCLEIC_SESSION_ID", "sess-1")
        // No `NUCLEIC_REAL_BASH` here — nash no longer honors it (docs/NASH.md §4.3) and picks
        // the target from the trusted policy or its compiled-in defaults, of which `/bin/bash`
        // is one.
        .env_remove("NUCLEIC_SHELL_PARENT")
        .status()
        .unwrap();
    // bash also rejects this input; what matters is bash's diagnostic + code.
    assert_eq!(status.code(), Some(2));

    let events = sink.events_until(|evs| find(evs, "fallback", |_| true).is_some());
    let fb = find(&events, "fallback", |_| true).expect("fallback event");
    assert_eq!(fb["input"], "if then fi");
    assert!(fb["reason"].as_str().unwrap().starts_with("parse-error"));
}

#[test]
fn silent_without_config() {
    // No NUCLEIC_SHELL_* config: nash must run normally and post nothing.
    let out = Command::new(env!("CARGO_BIN_EXE_nash"))
        .args(["-c", "echo quiet"])
        .env_remove("NUCLEIC_SHELL_SOCKET")
        .env_remove("NUCLEIC_SHELL_HOOK_URL")
        .env_remove("NUCLEIC_SHELL_SPOOL")
        .output()
        .unwrap();
    assert_eq!(out.status.code(), Some(0));
    assert_eq!(String::from_utf8_lossy(&out.stdout), "quiet\n");
}

fn decode_preview(event: &serde_json::Value) -> String {
    use std::io::Read;
    let b64 = event["previewB64"].as_str().unwrap_or("");
    // Minimal base64 decode for test assertions.
    let mut table = [255u8; 256];
    for (i, c) in b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/"
        .iter()
        .enumerate()
    {
        table[*c as usize] = i as u8;
    }
    let mut bits = 0u32;
    let mut nbits = 0;
    let mut out = Vec::new();
    for &c in b64.as_bytes() {
        if c == b'=' {
            break;
        }
        let v = table[c as usize];
        if v == 255 {
            continue;
        }
        bits = (bits << 6) | u32::from(v);
        nbits += 6;
        if nbits >= 8 {
            nbits -= 8;
            out.push((bits >> nbits) as u8);
        }
    }
    let mut s = String::new();
    let _ = out.as_slice().read_to_string(&mut s);
    s
}

#[test]
fn redirect_readback_captures_each_operator() {
    let sink = Sink::start();
    let status = run_nash(
        &sink,
        "echo first > d.txt; echo second >> d.txt; wc -c < d.txt > /dev/null",
    );
    assert_eq!(status.code(), Some(0));

    let events = sink.events_until(|evs| {
        evs.iter().filter(|e| e["kind"] == "redirect").count() >= 4
    });
    let redirs: Vec<_> = events.iter().filter(|e| e["kind"] == "redirect").collect();

    let trunc = redirs.iter().find(|e| e["op"] == ">").expect("truncate redirect");
    assert_eq!(decode_preview(trunc), "first\n");
    assert_eq!(trunc["bytes"], 6);
    // The target is reported ABSOLUTE (docs/NASH.md §5.2). A redirect event carries no cwd of its
    // own, and the host maps the path to a worktree file — and therefore to a file lock — so a bare
    // `d.txt` would be unresolvable there. `run_nash` runs in a fresh temp workspace.
    let target = trunc["target"].as_str().expect("redirect target");
    assert!(target.starts_with('/'), "target not absolute: {target}");
    assert!(target.ends_with("/d.txt"), "target lost its file name: {target}");

    let append = redirs.iter().find(|e| e["op"] == ">>").expect("append redirect");
    // Append captures ONLY the added bytes, not the whole file.
    assert_eq!(decode_preview(append), "second\n");
    assert_eq!(append["bytes"], 7);

    let input = redirs.iter().find(|e| e["op"] == "<").expect("input redirect");
    assert_eq!(decode_preview(input), "first\nsecond\n");

    // The /dev/null output redirect is metadata-only.
    let devnull = redirs
        .iter()
        .find(|e| e["target"] == "/dev/null")
        .expect("devnull redirect");
    assert_eq!(devnull["bytes"], 0);
    assert_eq!(devnull["previewB64"], "");
}

#[test]
fn heredoc_and_herestring_captured_inline() {
    let sink = Sink::start();
    let status = run_nash(&sink, "cat <<EOF > /dev/null\nline one\nline two\nEOF\ncat <<< 'here string body' > /dev/null");
    assert_eq!(status.code(), Some(0));

    let events = sink.events_until(|evs| {
        evs.iter().filter(|e| e["kind"] == "redirect" && (e["op"] == "<<" || e["op"] == "<<<")).count() >= 2
    });
    let here = events.iter().find(|e| e["op"] == "<<").expect("heredoc redirect");
    assert_eq!(decode_preview(here), "line one\nline two\n");
    let hstr = events.iter().find(|e| e["op"] == "<<<").expect("herestring redirect");
    assert_eq!(decode_preview(hstr), "here string body\n");
}

#[test]
fn cmdsub_output_captured() {
    let sink = Sink::start();
    let status = run_nash(&sink, "x=$(printf 'sub output'); echo \"$x\" > /dev/null");
    assert_eq!(status.code(), Some(0));
    let events = sink.events_until(|evs| evs.iter().any(|e| e["kind"] == "cmdsub"));
    let cs = events.iter().find(|e| e["kind"] == "cmdsub").expect("cmdsub event");
    assert_eq!(decode_preview(cs), "sub output");
    assert_eq!(cs["bytes"], 10);
}

#[test]
fn credential_shapes_redacted_in_previews() {
    let sink = Sink::start();
    let status = run_nash(
        &sink,
        "echo 'export TOKEN=ghp_ABCDEFGHIJKLMNOP1234567890abcd' > creds.txt",
    );
    assert_eq!(status.code(), Some(0));
    let events = sink.events_until(|evs| evs.iter().any(|e| e["kind"] == "redirect"));
    let r = events.iter().find(|e| e["kind"] == "redirect").expect("redirect");
    let preview = decode_preview(r);
    assert!(!preview.contains("ghp_ABCDEFGHIJKLMNOP"), "token must be redacted: {preview:?}");
    assert!(preview.contains("«redacted»"), "expected redaction marker: {preview:?}");
    // The byte count still reflects the real (unredacted) length on disk.
    assert_eq!(r["bytes"], 48);
}

#[test]
fn pipe_links_captured_with_indices() {
    let sink = Sink::start();
    let status = run_nash(&sink, "printf 'a\\nb\\nc\\n' | grep -n . | cat > /dev/null");
    assert_eq!(status.code(), Some(0));

    let events = sink.events_until(|evs| {
        evs.iter().filter(|e| e["kind"] == "pipe").count() >= 2
    });
    let pipes: Vec<_> = events.iter().filter(|e| e["kind"] == "pipe").collect();
    assert_eq!(pipes.len(), 2, "two links in a three-stage pipeline");
    // All links share one pipeline id.
    assert_eq!(pipes[0]["pipelineId"], pipes[1]["pipelineId"]);

    let link0 = pipes.iter().find(|e| e["fromIndex"] == 0).expect("link 0->1");
    assert_eq!(link0["toIndex"], 1);
    assert_eq!(decode_preview(link0), "a\nb\nc\n");
    assert_eq!(link0["bytes"], 6);

    let link1 = pipes.iter().find(|e| e["fromIndex"] == 1).expect("link 1->2");
    assert_eq!(link1["toIndex"], 2);
    assert_eq!(decode_preview(link1), "1:a\n2:b\n3:c\n");
}

/// Each stage's exec event carries the slot that joins it to the links either side, so a viewer
/// can name a link's producer and consumer (`printf … | grep …`) instead of showing bare indices.
#[test]
fn pipeline_stages_carry_their_slot_on_exec() {
    let sink = Sink::start();
    let status = run_nash(&sink, "printf 'a\\nb\\n' | grep -n . | cat > /dev/null");
    assert_eq!(status.code(), Some(0));

    let events = sink.events_until(|evs| {
        ["printf", "grep", "cat"]
            .iter()
            .all(|c| find(evs, "exec", |e| &e["argv"][0] == c).is_some())
            && evs.iter().any(|e| e["kind"] == "pipe")
    });

    let slot = |cmd: &str| {
        find(&events, "exec", |e| &e["argv"][0] == cmd)
            .unwrap_or_else(|| panic!("{cmd} exec event missing"))["pipeline"]
            .clone()
    };
    for (cmd, index) in [("printf", 0), ("grep", 1), ("cat", 2)] {
        assert_eq!(slot(cmd)["index"], index, "{cmd} is stage {index}");
        assert_eq!(slot(cmd)["len"], 3, "{cmd} sees a three-stage pipeline");
    }
    // The stages and the links they connect share one id — that join is the whole point.
    let pipeline_id = slot("printf")["id"].clone();
    assert!(pipeline_id.is_number());
    assert_eq!(slot("grep")["id"], pipeline_id);
    assert_eq!(slot("cat")["id"], pipeline_id);
    let pipe = find(&events, "pipe", |_| true).expect("pipe event");
    assert_eq!(pipe["pipelineId"], pipeline_id);
}

/// A standalone command has no slot at all, so the field stays absent rather than reporting a
/// meaningless stage 0 of 1.
#[test]
fn standalone_command_carries_no_pipeline_slot() {
    let sink = Sink::start();
    assert_eq!(run_nash(&sink, "echo alone").code(), Some(0));

    let events = sink.events_until(|evs| find(evs, "exec", |e| e["argv"][0] == "echo").is_some());
    let echo = find(&events, "exec", |e| e["argv"][0] == "echo").expect("echo exec event");
    assert!(echo.get("pipeline").is_none(), "no pipeline field on a lone command");
}

/// A command substitution runs during a stage's *expansion*, before the stage's own command, so
/// it must not claim the stage — otherwise `grep "$(printf x)" … | wc -l` reports `printf x` as
/// what produced the piped bytes.
#[test]
fn command_substitution_does_not_claim_its_stage() {
    let sink = Sink::start();
    let status = run_nash(&sink, "grep \"$(printf a)\" /etc/hostname | cat > /dev/null");
    assert_eq!(status.code(), Some(0));

    let events = sink.events_until(|evs| {
        find(evs, "exec", |e| e["argv"][0] == "grep").is_some()
            && find(evs, "exec", |e| e["argv"][0] == "cat").is_some()
    });
    let printf = find(&events, "exec", |e| e["argv"][0] == "printf").expect("printf exec event");
    assert!(
        printf.get("pipeline").is_none(),
        "the substituted command is an argument to stage 0, not stage 0"
    );
    let grep = find(&events, "exec", |e| e["argv"][0] == "grep").expect("grep exec event");
    assert_eq!(grep["pipeline"]["index"], 0, "grep is what stage 0 actually ran");
}

/// A stage that is a compound command attributes the commands *inside* it to that same stage —
/// they really are what that stage ran — while a nested pipeline takes over with its own id.
#[test]
fn nested_commands_inherit_their_enclosing_stage() {
    let sink = Sink::start();
    let status = run_nash(&sink, "{ echo one; echo two; } | cat > /dev/null");
    assert_eq!(status.code(), Some(0));

    let events = sink.events_until(|evs| {
        find(evs, "exec", |e| e["argv"][1] == "two").is_some()
            && find(evs, "exec", |e| e["argv"][0] == "cat").is_some()
    });
    for arg in ["one", "two"] {
        let ev = find(&events, "exec", |e| e["argv"][1] == arg).expect("echo exec event");
        assert_eq!(ev["pipeline"]["index"], 0, "`echo {arg}` runs inside stage 0");
    }
    let cat = find(&events, "exec", |e| e["argv"][0] == "cat").expect("cat exec event");
    assert_eq!(cat["pipeline"]["index"], 1);
}

/// Each link names the commands either side of it, rendered from their AST at spawn. This is what
/// makes `curl … | sh` legible; deriving it from exec events cannot, because a compound stage emits
/// one exec per command inside it and none of them says which owned the pipe.
#[test]
fn pipe_links_carry_their_stage_text() {
    let sink = Sink::start();
    let status = run_nash(&sink, "printf 'a\\nb\\n' | grep -v a | cat > /dev/null");
    assert_eq!(status.code(), Some(0));

    let events = sink.events_until(|evs| evs.iter().filter(|e| e["kind"] == "pipe").count() >= 2);
    let pipes: Vec<_> = events.iter().filter(|e| e["kind"] == "pipe").collect();
    let link0 = pipes.iter().find(|e| e["fromIndex"] == 0).expect("link 0->1");
    // Quoting is preserved as written, and a stage's redirections are part of the stage.
    assert_eq!(link0["fromText"], "printf 'a\\nb\\n'");
    assert_eq!(link0["toText"], "grep -v a");
    let link1 = pipes.iter().find(|e| e["fromIndex"] == 1).expect("link 1->2");
    assert_eq!(link1["fromText"], "grep -v a");
    assert_eq!(link1["toText"], "cat > /dev/null");
}

/// The case exec-derived naming cannot get right: the stage is a compound command, so several exec
/// events share its slot and none of them is "the" stage.
#[test]
fn compound_stage_is_named_as_a_whole() {
    let sink = Sink::start();
    let status = run_nash(&sink, "{ echo one; echo two; } | cat > /dev/null");
    assert_eq!(status.code(), Some(0));

    let events = sink.events_until(|evs| evs.iter().any(|e| e["kind"] == "pipe"));
    let pipe = find(&events, "pipe", |_| true).expect("pipe event");
    let from = pipe["fromText"].as_str().unwrap();
    assert!(from.contains("echo one"), "names the whole stage, got {from:?}");
    assert!(from.contains("echo two"), "names the whole stage, got {from:?}");
    assert_eq!(pipe["toText"], "cat > /dev/null");
}

/// Stage text is source, so it can carry a literal credential exactly as argv can. It goes through
/// the same in-guest masking as every other outbound string (docs/NASH.md §5.4).
#[test]
fn stage_text_is_redacted() {
    let sink = Sink::start();
    let status = run_nash(
        &sink,
        "echo 'Bearer ghp_abcdefghijklmnopqrstuvwxyz012345' | cat > /dev/null",
    );
    assert_eq!(status.code(), Some(0));

    let events = sink.events_until(|evs| evs.iter().any(|e| e["kind"] == "pipe"));
    let pipe = find(&events, "pipe", |_| true).expect("pipe event");
    let from = pipe["fromText"].as_str().unwrap();
    assert!(!from.contains("ghp_abcdefghij"), "credential must not survive: {from:?}");
    assert!(from.contains("redacted"), "and must be visibly masked: {from:?}");
}

/// A credential that straddles the stage-label cap must still be masked. The cap runs in brush-core
/// and appends `…` to the surviving text; glued onto the final token, that marker used to disqualify
/// it from the redactor's high-entropy branch and walk a truncated secret out of the guest.
#[test]
fn stage_text_redacts_a_secret_split_by_the_cap() {
    let sink = Sink::start();
    // Land the key astride the 200-char cap with enough of it left to still look high-entropy:
    // `echo -d <pad> -d ` is 164 chars, so 36 of the key's 40 survive and pick up the `…`.
    let secret = "wJalrXUtnFEMIK7MDENGbPxRfiCYEXAMPLEKEY01";
    let pad = "a".repeat(152);
    // `PASTTHECAP` sits wholly beyond 200 chars, so its absence proves the cap actually fired —
    // the `…` marker cannot prove it here, since redacting the straddling token consumes it.
    let status = run_nash(
        &sink,
        &format!("echo -d {pad} -d {secret} PASTTHECAP | cat > /dev/null"),
    );
    assert_eq!(status.code(), Some(0));

    let events = sink.events_until(|evs| evs.iter().any(|e| e["kind"] == "pipe"));
    let pipe = find(&events, "pipe", |_| true).expect("pipe event");
    let from = pipe["fromText"].as_str().unwrap();
    assert!(
        !from.contains("PASTTHECAP"),
        "the label must actually have been capped for this to test anything: {from:?}"
    );
    assert!(
        !from.contains(&secret[..32]),
        "a credential split by the cap must not survive: {from:?}"
    );
    assert!(from.contains("redacted"), "and must be visibly masked: {from:?}");
}

/// A stage can be an entire loop body; the label is one line and bounded, because it rides on every
/// link of every pipeline.
#[test]
fn stage_text_is_single_line_and_capped() {
    let sink = Sink::start();
    let long = "x".repeat(400);
    let status = run_nash(
        &sink,
        &format!("while read l; do echo \"$l{long}\"; done < /etc/hostname | cat > /dev/null"),
    );
    assert_eq!(status.code(), Some(0));

    let events = sink.events_until(|evs| evs.iter().any(|e| e["kind"] == "pipe"));
    let pipe = find(&events, "pipe", |_| true).expect("pipe event");
    let from = pipe["fromText"].as_str().unwrap();
    assert!(!from.contains('\n'), "a label is one line: {from:?}");
    assert!(from.chars().count() <= 201, "capped, got {} chars", from.chars().count());
    assert!(from.ends_with('…'), "and marked as elided: {from:?}");
}

#[test]
fn pipe_sigpipe_consumer_exits_early() {
    // `yes | head` — the consumer closes after N lines; the producer must get
    // SIGPIPE/EPIPE and the whole thing must terminate (not hang), with the link
    // still reported.
    let sink = Sink::start();
    let status = run_nash(&sink, "yes | head -3 > /dev/null");
    assert_eq!(status.code(), Some(0));
    let events = sink.events_until(|evs| evs.iter().any(|e| e["kind"] == "pipe"));
    assert!(events.iter().any(|e| e["kind"] == "pipe"), "link reported despite early consumer exit");
}

#[test]
fn pipe_preserves_binary_stream() {
    // Byte-for-byte integrity across a tapped pipe: md5 through nash must match bash.
    let dir = scratch_dir("nash-pipebin");
    let src = dir.join("rand.bin");
    let data: Vec<u8> = (0..100_000u32).map(|i| (i.wrapping_mul(2654435761) >> 16) as u8).collect();
    std::fs::write(&src, &data).unwrap();
    let script = format!("cat {} | cat | md5sum", src.display());

    let sink = Sink::start();
    let nash_out = Command::new(env!("CARGO_BIN_EXE_nash"))
        .args(["-c", &script])
        .env("NUCLEIC_SHELL_SOCKET", sink.socket())
        .env("NUCLEIC_SESSION_ID", "sess-1")
        .env_remove("NUCLEIC_SHELL_PARENT")
        .output()
        .unwrap();
    let bash_out = Command::new("/bin/bash").args(["-c", &script]).output().unwrap();
    assert_eq!(nash_out.stdout, bash_out.stdout, "md5 through tapped pipe must match bash");
    let _ = std::fs::remove_dir_all(&dir);
}

#[test]
fn redirect_preserves_seek_semantics() {
    // A seeking writer (dd with seek=) must see a real, seekable fd — the file
    // must end up byte-identical to bash, proving nash didn't interpose a pipe.
    let sink = Sink::start();
    let parent = scratch_dir("nash-seek");
    let script = "printf '0123456789' > f.bin; dd if=/dev/zero of=f.bin bs=1 seek=3 count=2 conv=notrunc 2>/dev/null; od -An -tx1 f.bin";
    let out = Command::new(env!("CARGO_BIN_EXE_nash"))
        .args(["-c", script])
        .current_dir(&parent)
        .env("NUCLEIC_SHELL_SOCKET", sink.socket())
        .env("NUCLEIC_HOOK_TOKEN", "test-token")
        .env("NUCLEIC_SESSION_ID", "sess-1")
        .env_remove("NUCLEIC_SHELL_PARENT")
        .output()
        .unwrap();
    assert_eq!(out.status.code(), Some(0));
    // Bytes 3 and 4 zeroed by the seeking write; the rest intact.
    let text = String::from_utf8_lossy(&out.stdout);
    let hex: String = text.split_whitespace().collect::<Vec<_>>().join(" ");
    assert_eq!(hex, "30 31 32 00 00 35 36 37 38 39");
    let _ = std::fs::remove_dir_all(&parent);
}

#[test]
fn spool_fallback_when_socket_absent() {
    let dir = scratch_dir("nash-spool");
    let status = Command::new(env!("CARGO_BIN_EXE_nash"))
        .args(["-c", "echo spooled"])
        .env("NUCLEIC_SHELL_SOCKET", "/nonexistent/control.sock")
        .env("NUCLEIC_SHELL_SPOOL", &dir)
        .env("NUCLEIC_SESSION_ID", "sess-1")
        .env_remove("NUCLEIC_SHELL_HOOK_URL")
        .env_remove("NUCLEIC_SHELL_PARENT")
        .status()
        .unwrap();
    assert_eq!(status.code(), Some(0));

    let mut found = false;
    for entry in std::fs::read_dir(&dir).expect("spool dir created") {
        let content = std::fs::read_to_string(entry.unwrap().path()).unwrap();
        for line in content.lines() {
            let batch: serde_json::Value = serde_json::from_str(line).unwrap();
            if batch["events"]
                .as_array()
                .unwrap()
                .iter()
                .any(|e| e["kind"] == "exec" && e["argv"][0] == "echo")
            {
                found = true;
            }
        }
    }
    assert!(found, "exec event spooled to disk");
    let _ = std::fs::remove_dir_all(&dir);
}

/// The bypass this mechanism exists to close (docs/NASH.md §4.3): the agent sets the kill
/// switch in its own environment. nash must run the command *as nash* anyway — the exec event
/// proves it did, since a re-exec'd bash posts nothing — and must report the attempt.
#[test]
fn env_kill_switch_is_ignored_and_reported() {
    let sink = Sink::start();
    let status = Command::new(env!("CARGO_BIN_EXE_nash"))
        .args(["-c", "echo still-observed"])
        .env("NUCLEIC_SHELL_SOCKET", sink.socket())
        .env("NUCLEIC_HOOK_TOKEN", "test-token")
        .env("NUCLEIC_SESSION_ID", "sess-1")
        .env("NUCLEIC_NASH_DISABLE", "1")
        .env("NUCLEIC_REAL_BASH", "/bin/sh")
        .env_remove("NUCLEIC_SHELL_PARENT")
        .status()
        .unwrap();
    assert_eq!(status.code(), Some(0));

    let events = sink.events_until(|evs| {
        find(evs, "exec", |e| e["argv"][0] == "echo").is_some()
            && find(evs, "policy", |e| e["lever"] == "NUCLEIC_REAL_BASH").is_some()
    });

    // Observation survived the attempt.
    assert!(find(&events, "exec", |e| e["argv"][0] == "echo").is_some());

    let disable = find(&events, "policy", |e| e["lever"] == "NUCLEIC_NASH_DISABLE")
        .expect("policy event for the kill switch");
    assert_eq!(disable["action"], "ignored");
    assert_eq!(disable["value"], "1");

    let real_bash = find(&events, "policy", |e| e["lever"] == "NUCLEIC_REAL_BASH")
        .expect("policy event for the fallback target");
    assert_eq!(real_bash["action"], "ignored");
}

/// The quieter half of the same hole: rather than leaving, the agent blinds nash by telling it
/// not to capture. With no trusted policy nash has no way to know better and stays silent —
/// asserted here so `require_observation`'s baseline is documented rather than assumed. The
/// policy-on side needs a root-owned file, so it is the image smoke test that covers it.
#[test]
fn capture_off_silences_nash_without_a_policy() {
    let sink = Sink::start();
    let status = Command::new(env!("CARGO_BIN_EXE_nash"))
        .args(["-c", "echo unobserved"])
        .env("NUCLEIC_SHELL_SOCKET", sink.socket())
        .env("NUCLEIC_HOOK_TOKEN", "test-token")
        .env("NUCLEIC_SESSION_ID", "sess-1")
        .env("NUCLEIC_SHELL_CAPTURE", "off")
        .env_remove("NUCLEIC_SHELL_PARENT")
        .status()
        .unwrap();
    assert_eq!(status.code(), Some(0));
    assert!(
        sink.rx.recv_timeout(Duration::from_millis(750)).is_err(),
        "capture=off with no policy posts nothing"
    );
}

/// A shell posts a batch every 500 ms for as long as it lives, so the connection is opened once
/// and kept (docs/NASH_STREAM_PERF_PLAN.md §P4.1) rather than reconnected per batch. Two commands
/// either side of a flush window are two batches; one connection has to carry both.
#[test]
fn one_connection_carries_every_batch() {
    let sink = Sink::start();
    let status = run_nash(&sink, "echo one; sleep 0.9; echo two");
    assert_eq!(status.code(), Some(0));

    let events = sink.events_until(|evs| {
        find(evs, "exec", |e| e["argv"][0] == "echo" && e["argv"][1] == "two").is_some()
    });
    // Both commands were reported…
    assert!(find(&events, "exec", |e| e["argv"][1] == "one").is_some());
    assert!(find(&events, "exec", |e| e["argv"][1] == "two").is_some());
    // …and the shell only ever dialled the sink once.
    assert_eq!(
        sink.connections.load(Ordering::Relaxed),
        1,
        "batches must share one kept-open connection"
    );
}

/// The budget's end-to-end half (docs/NASH.md §5.4): a burst of fat pipes in one batch keeps every
/// event and every byte count, and gives up only the previews past the allowance.
#[test]
fn batch_preview_budget_keeps_counts_and_drops_previews() {
    let sink = Sink::start();
    // 24 links × 64 KiB of payload each — comfortably past the 1 MiB per-batch preview budget,
    // and fast enough that they all land in one 500 ms flush.
    let status = run_nash(
        &sink,
        "for i in $(seq 1 24); do head -c 65536 /dev/zero | cat > /dev/null; done",
    );
    assert_eq!(status.code(), Some(0));

    let events = sink.events_until(|evs| evs.iter().filter(|e| e["kind"] == "pipe").count() >= 24);
    let pipes: Vec<_> = events.iter().filter(|e| e["kind"] == "pipe").collect();
    assert!(pipes.len() >= 24, "every link is still reported: {}", pipes.len());
    // Nothing loses its count or its fingerprint…
    for pipe in &pipes {
        assert_eq!(pipe["bytes"], 65536);
        assert_ne!(pipe["hash"], "");
    }
    // …and the ones past the budget are the ones without a preview.
    let dropped = pipes.iter().filter(|p| p["previewB64"] == "").count();
    assert!(dropped > 0, "the budget dropped no previews");
    assert!(dropped < pipes.len(), "the budget dropped every preview");
}
