cloudfox-coreshift-core 2.29.1

// This Source Code Form is subject to the terms of the Mozilla Public
// License, v. 2.0. If a copy of the MPL was not distributed with this
// file, You can obtain one at https://mozilla.org/MPL/2.0/

use crate::fd::Fd;
use crate::fs::{mmap_madvise, path_exists, path_lstat_exists, readahead};
use crate::inotify::{InotifyEvent, decode_events};
use crate::io::{DrainState, SinkResult};
use crate::proc::{
    ProcDir, clock_ticks_per_second, parse_proc_cmdline_bytes, parse_proc_status, path_lstat,
    path_stat, path_uid, read_proc_cmdline_at, stat_at, uid, uid_at,
};
use crate::reactor::Reactor;
use crate::signal::{install_shutdown_flag, install_shutdown_flag_guard, shutdown_requested};
use crate::socket::{
    StaleSocketPolicy, UnixConnectResult, UnixSocketAddr, UnixSocketBindOptions, bind, chmod,
    connect,
};
use crate::spawn::{
    CancelPolicy, ExitStatus, Process, ProcessGroup, SessionExitPolicy, SpawnBackend,
    SpawnFdPolicy, SpawnOptions, SpawnOptionsBuilder, make_pty, orphan_session,
    pty_session_enumerated, pty_window, reset_pty_enumeration, spawn_managed, spawn_start,
};
use std::fs::{File, remove_file};
use std::io::Write;
use std::os::unix::ffi::OsStringExt;
use std::os::unix::fs::{FileTypeExt, PermissionsExt};
use std::os::unix::io::{AsRawFd, RawFd};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};

fn with_temp_readahead_file<T>(f: impl FnOnce(File, &std::path::Path) -> T) -> T {
    let path = std::env::temp_dir().join(format!(
        "coreshift_test_readahead_{}_{}",
        std::process::id(),
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_nanos()
    ));

    let mut file = File::options()
        .read(true)
        .write(true)
        .create(true)
        .truncate(true)
        .open(&path)
        .unwrap();
    file.write_all(b"readahead test data").unwrap();
    file.sync_all().unwrap();

    let result = f(file, &path);
    let _ = remove_file(&path);
    result
}

fn temp_socket_path(name: &str) -> PathBuf {
    std::env::temp_dir().join(format!(
        "coreshift_test_{name}_{}_{}",
        std::process::id(),
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_nanos()
    ))
}

fn bind_test_unix_listener(path: &Path, unlink_stale: bool) -> crate::socket::UnixListener {
    bind(
        UnixSocketAddr::Path(path),
        UnixSocketBindOptions {
            stale_socket_policy: if unlink_stale {
                StaleSocketPolicy::UnlinkSocketOnly
            } else {
                StaleSocketPolicy::Preserve
            },
            mode: None,
        },
    )
    .unwrap()
}

fn connect_test_unix_stream(addr: UnixSocketAddr<'_>) -> crate::socket::UnixStream {
    match connect(addr).unwrap() {
        UnixConnectResult::Connected(stream) => stream,
        UnixConnectResult::InProgress(stream) => stream.finish_connect().unwrap(),
    }
}

fn assert_readahead_result(result: Result<(), crate::CoreError>) {
    match result {
        Ok(()) => {}
        Err(err) if err.raw_os_error() == Some(libc::ENOSYS) => {
            eprintln!("skipping readahead test: unsupported on this target");
        }
        Err(err) => panic!("readahead failed unexpectedly: {err}"),
    }
}

struct RawFdRef(RawFd);

static TEST_SHUTDOWN_FLAG_A: AtomicBool = AtomicBool::new(false);
static TEST_SHUTDOWN_FLAG_B: AtomicBool = AtomicBool::new(false);

// The shutdown-flag tests below install/restore *process-global* SIGINT/SIGTERM
// dispositions and the global shutdown flag pointer. Cargo runs tests in
// parallel threads, so without serialization two of them can interleave and
// clobber each other's handlers (e.g. an install test replacing SIG_IGN with
// the shutdown handler between the guard test's drop and its final assert).
// Serialize every test that touches global signal state behind this lock.
static SIGNAL_TEST_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(());

impl AsRawFd for RawFdRef {
    fn as_raw_fd(&self) -> RawFd {
        self.0
    }
}

#[test]
fn test_log_backend_selection() {
    use crate::log::{LogBackend, LogLevel, Logger};

    // We can't easily capture stderr/android output here without more complex setup,
    // but we can verify that different backends work.
    let logger_null = Logger::new(LogBackend::Null);
    logger_null.log(LogLevel::Info, "Test", "This should be discarded");

    let logger_stderr = Logger::new(LogBackend::Stderr);
    logger_stderr.log(LogLevel::Info, "Test", "This should go to stderr");
}

#[test]
#[ignore]
fn test_signalfd_new() {
    use crate::reactor::Reactor;
    use crate::signal::{SIGUSR1, SignalRuntime};

    let signals = SignalRuntime::set_with(&[SIGUSR1]).unwrap();

    // Block the signal so it can be read from signalfd
    let old_mask = SignalRuntime::block_current_thread(&signals).unwrap();

    let sfd = SignalRuntime::signalfd_new(&signals).unwrap();
    let mut reactor = Reactor::new().unwrap();
    let token = reactor.add(&sfd, true, false).unwrap();

    // Send signal to self
    unsafe {
        libc::kill(libc::getpid(), SIGUSR1);
    }

    let mut events = Vec::new();
    let n = reactor.wait(&mut events, 1, 1000).unwrap();
    assert_eq!(n, 1);
    assert_eq!(events[0].token, token);
    assert!(events[0].readable);

    // Clean up
    SignalRuntime::restore_current_thread(&old_mask).unwrap();
}

#[test]
fn test_decode_inotify_events() {
    // Mock multiple inotify_event records.
    let mut buf = Vec::new();

    // Event 1: wd=1, mask=0x2, len=0
    buf.extend_from_slice(&1i32.to_ne_bytes());
    buf.extend_from_slice(&2u32.to_ne_bytes());
    buf.extend_from_slice(&0u32.to_ne_bytes()); // cookie
    buf.extend_from_slice(&0u32.to_ne_bytes()); // len

    // Event 2: wd=2, mask=0x4, len=8 (with name padding)
    buf.extend_from_slice(&2i32.to_ne_bytes());
    buf.extend_from_slice(&4u32.to_ne_bytes());
    buf.extend_from_slice(&0u32.to_ne_bytes()); // cookie
    buf.extend_from_slice(&8u32.to_ne_bytes()); // len
    buf.extend_from_slice(b"file.txt"); // 8 bytes

    let events = decode_events(&buf).unwrap();
    assert_eq!(events.len(), 2);
    assert_eq!(
        events[0],
        InotifyEvent {
            wd: 1,
            mask: 2,
            name: None,
        }
    );
    assert_eq!(
        events[1],
        InotifyEvent {
            wd: 2,
            mask: 4,
            name: Some(b"file.txt".to_vec()),
        }
    );

    // Event 3: truncated (only 8 bytes of header)
    buf.extend_from_slice(&3i32.to_ne_bytes());
    buf.extend_from_slice(&8u32.to_ne_bytes());

    let err = decode_events(&buf).unwrap_err();
    assert_eq!(err.raw_os_error(), Some(libc::EINVAL));
}

#[test]
fn test_parse_proc_status() {
    let content = "Name:\tcore_daemon\nState:\tR (running)\nUid:\t1000\t1000\t1000\t1000\nGid:\t1000\t1000\t1000\t1000\n";
    let status = parse_proc_status(content).unwrap();
    assert_eq!(status.name, "core_daemon");
    assert_eq!(status.uid, 1000);
}

#[test]
fn test_parse_proc_status_missing_uid_is_invalid() {
    let content = "Name:\tcore_daemon\nState:\tR (running)\n";
    let err = parse_proc_status(content).unwrap_err();
    assert_eq!(err.raw_os_error(), Some(libc::EINVAL));
}

#[test]
fn test_parse_proc_cmdline_bytes_replaces_nul_separators() {
    let content = b"/system/bin/sh\0-c\0echo hello\0";
    assert_eq!(
        parse_proc_cmdline_bytes(content),
        "/system/bin/sh -c echo hello"
    );
}

#[test]
fn test_read_proc_cmdline_at_uses_explicit_root() {
    let proc_root = std::env::temp_dir().join(format!(
        "coreshift_test_proc_root_{}_{}",
        std::process::id(),
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_nanos()
    ));
    let pid_dir = proc_root.join("12345");
    std::fs::create_dir_all(&pid_dir).unwrap();
    std::fs::write(pid_dir.join("cmdline"), b"com.example.app\0:service\0").unwrap();

    let cmdline = read_proc_cmdline_at(&proc_root, 12345).unwrap();
    assert_eq!(cmdline, "com.example.app :service");

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

#[test]
fn test_proc_dir_pins_process_and_reads_via_dirfd() {
    let proc_root = std::env::temp_dir().join(format!(
        "coreshift_test_proc_dir_{}_{}",
        std::process::id(),
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_nanos()
    ));
    let pid_dir = proc_root.join("4242");
    std::fs::create_dir_all(&pid_dir).unwrap();
    std::fs::write(pid_dir.join("cmdline"), b"com.example.manager\0").unwrap();
    std::fs::write(pid_dir.join("status"), b"Name:\tmanager\nUid:\t12345\n").unwrap();

    let dir = ProcDir::open_at(&proc_root, 4242).unwrap();
    assert_eq!(dir.cmdline().unwrap(), "com.example.manager");
    let status = dir.status().unwrap();
    assert_eq!(status.name, "manager");
    assert_eq!(status.uid, 12345);
    // The pinned dirfd's stat reflects the same directory the reads came from:
    // for a fake root the dir's real owner (the test runner) is what `stat_at`
    // reports; on real procfs `/proc/<pid>` is owned by the process's uid.
    assert_eq!(
        dir.stat().unwrap().uid,
        stat_at(&proc_root, 4242).unwrap().uid
    );

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

#[test]
fn test_exec_context_validation() {
    // Empty argv
    let res = SpawnOptions::builder(vec![], SpawnBackend::Fork).build();
    assert!(res.is_err());

    // Interior NUL in argv
    let res = SpawnOptions::builder(
        vec!["valid".to_string(), "inv\0alid".to_string()],
        SpawnBackend::Fork,
    )
    .build();
    assert!(res.is_err());

    // Valid
    let res = SpawnOptions::builder(
        vec!["/bin/ls".to_string(), "-l".to_string()],
        SpawnBackend::Fork,
    )
    .cwd("/tmp".to_string())
    .build();
    assert!(res.is_ok());
}

#[test]
fn test_posix_spawn_rejects_unsupported_cwd() {
    let opts = SpawnOptions::builder(vec!["/bin/true".to_string()], SpawnBackend::PosixSpawn)
        .cwd("/tmp".to_string())
        .build()
        .unwrap();

    let err = match spawn_start(opts) {
        Ok(_) => panic!("expected unsupported posix_spawn cwd to fail"),
        Err(err) => err,
    };
    assert_eq!(err.raw_os_error(), Some(libc::EINVAL));
}

#[test]
fn test_posix_spawn_rejects_unsupported_setsid() {
    let opts = SpawnOptions::builder(vec!["/bin/true".to_string()], SpawnBackend::PosixSpawn)
        .pgroup(ProcessGroup::new(None, true))
        .build()
        .unwrap();

    let err = match spawn_start(opts) {
        Ok(_) => panic!("expected unsupported posix_spawn setsid to fail"),
        Err(err) => err,
    };
    assert_eq!(err.raw_os_error(), Some(libc::EINVAL));
}

#[test]
fn test_posix_spawn_rejects_close_from3_fd_policy() {
    let opts = SpawnOptions::builder(vec!["/bin/true".to_string()], SpawnBackend::PosixSpawn)
        .fd_policy(SpawnFdPolicy::CloseFrom3)
        .build()
        .unwrap();

    let err = match spawn_start(opts) {
        Ok(_) => panic!("expected unsupported fd policy to fail"),
        Err(err) => err,
    };
    assert_eq!(err.raw_os_error(), Some(libc::EINVAL));
}

#[test]
fn test_fork_reports_child_chdir_error() {
    let err = SpawnOptions::builder(vec!["/bin/true".to_string()], SpawnBackend::Fork)
        .cwd("/definitely/missing/coreshift-core-test".to_string())
        .build()
        .unwrap()
        .run()
        .unwrap_err();
    assert_eq!(err.raw_os_error(), Some(libc::ENOENT));
    assert!(err.to_string().contains("spawn child chdir"));
}

#[test]
fn test_spawn_fd_policy_rejects_dirty_allowlist() {
    let file = File::open("/dev/null").unwrap();
    let fd = file.as_raw_fd();
    let err = SpawnOptions::builder(vec!["/bin/true".to_string()], SpawnBackend::PosixSpawn)
        .fd_policy(SpawnFdPolicy::Allowlist(vec![fd, fd]))
        .build()
        .unwrap()
        .run()
        .unwrap_err();
    assert_eq!(err.raw_os_error(), Some(libc::EINVAL));
}

#[test]
fn test_fork_close_from3_closes_inherited_fd() {
    let file = File::open("/dev/null").unwrap();
    let fd = file.as_raw_fd();
    let flags = unsafe { libc::fcntl(fd, libc::F_GETFD) };
    assert!(flags >= 0);
    let ret = unsafe { libc::fcntl(fd, libc::F_SETFD, flags & !libc::FD_CLOEXEC) };
    assert_eq!(ret, 0);

    let script = format!("if [ -e /proc/$$/fd/{fd} ]; then echo open; else echo closed; fi");
    let output = SpawnOptions::builder(
        vec!["/bin/sh".to_string(), "-c".to_string(), script],
        SpawnBackend::Fork,
    )
    .fd_policy(SpawnFdPolicy::CloseFrom3)
    .capture_stdout()
    .build()
    .unwrap()
    .run()
    .unwrap();

    assert_eq!(String::from_utf8_lossy(&output.stdout).trim(), "closed");
}

#[test]
fn test_exec_backends_support_isolated_pgroup_and_close_from3() {
    for backend in [
        SpawnBackend::Fork,
        SpawnBackend::Vfork,
        SpawnBackend::Clone3,
        SpawnBackend::Clone3Pidfd,
    ] {
        let opts = SpawnOptions::builder(vec!["/bin/true".to_string()], backend)
            .pgroup(ProcessGroup::new(None, true))
            .fd_policy(SpawnFdPolicy::CloseFrom3)
            .build()
            .unwrap();
        let running = spawn_start(opts).unwrap_or_else(|e| {
            panic!("{backend:?} should accept isolated pgroup + CloseFrom3: {e}")
        });
        let _ = running.process.wait_blocking();
    }
}

#[test]
fn test_exec_backends_reject_isolated_custom_leader() {
    for backend in [
        SpawnBackend::Fork,
        SpawnBackend::Vfork,
        SpawnBackend::Clone3,
        SpawnBackend::Clone3Pidfd,
    ] {
        let opts = SpawnOptions::builder(vec!["/bin/true".to_string()], backend)
            .pgroup(ProcessGroup::new(Some(5), true))
            .build()
            .unwrap();
        let err = match spawn_start(opts) {
            Ok(_) => panic!("{backend:?} must reject isolated + custom leader"),
            Err(e) => e,
        };
        assert_eq!(err.raw_os_error(), Some(libc::EINVAL));
    }
}

#[test]
fn test_posix_spawn_rejects_session_containment() {
    let opts = SpawnOptions::builder(vec!["/bin/true".to_string()], SpawnBackend::PosixSpawn)
        .pgroup(ProcessGroup::new(None, true))
        .session_containment()
        .build()
        .unwrap();

    let err = match spawn_start(opts) {
        Ok(_) => panic!("expected posix_spawn session containment to fail"),
        Err(err) => err,
    };
    assert_eq!(err.raw_os_error(), Some(libc::EINVAL));
}

#[test]
fn test_session_containment_requires_isolated_pgroup() {
    for backend in [
        SpawnBackend::Fork,
        SpawnBackend::Vfork,
        SpawnBackend::Clone3,
        SpawnBackend::Clone3Pidfd,
    ] {
        let opts = SpawnOptions::builder(vec!["/bin/true".to_string()], backend)
            .session_containment()
            .build()
            .unwrap();
        let err = match spawn_start(opts) {
            Ok(_) => panic!("{backend:?} must require isolated pgroup for containment"),
            Err(err) => err,
        };
        assert_eq!(err.raw_os_error(), Some(libc::EINVAL));
    }
}

#[test]
fn test_session_containment_allows_normal_exec() {
    for backend in [
        SpawnBackend::Fork,
        SpawnBackend::Vfork,
        SpawnBackend::Clone3,
        SpawnBackend::Clone3Pidfd,
    ] {
        let out = SpawnOptions::builder(
            vec!["/bin/echo".to_string(), "contained".to_string()],
            backend,
        )
        .pgroup(ProcessGroup::new(None, true))
        .session_containment()
        .capture_stdout()
        .build()
        .unwrap()
        .run()
        .unwrap();
        assert!(matches!(out.status, Some(ExitStatus::Exited(0))));
        assert_eq!(String::from_utf8_lossy(&out.stdout).trim(), "contained");
    }
}

#[test]
fn test_session_containment_blocks_setsid_escape() {
    // A child that tries to `setsid` out of its isolated session must fail
    // with EPERM, exit quickly, and never detach a survivor the daemon's
    // kill_group cannot reach. Without containment, `setsid sleep 60` would
    // succeed, escape to a new session, and hold the pipes open for 60s.
    let setsid = ["/usr/bin/setsid", "/bin/setsid", "/system/bin/setsid"]
        .into_iter()
        .find(|p| crate::fs::path_exists(p));
    let Some(setsid) = setsid else {
        return;
    };

    let started = std::time::Instant::now();
    let managed = spawn_managed(
        SpawnOptions::builder(
            vec![
                "/bin/sh".to_string(),
                "-c".to_string(),
                format!("{setsid} sleep 60"),
            ],
            SpawnBackend::Fork,
        )
        .pgroup(ProcessGroup::new(None, true))
        .session_containment()
        .capture_stderr()
        .timeout_ms(10_000)
        .kill_grace_ms(200)
        .cancel(CancelPolicy::Kill)
        .build()
        .unwrap(),
    )
    .unwrap();

    let output = drive_managed(managed, false).unwrap();
    assert!(
        !output.timed_out,
        "contained setsid must fail fast, not hang for the escaped sleep"
    );
    assert!(
        started.elapsed().as_secs() < 8,
        "contained setsid must exit promptly"
    );
    let stderr = String::from_utf8_lossy(&output.stderr);
    assert!(
        stderr.contains("setsid") && !stderr.is_empty(),
        "expected setsid utility to report the denied syscall, got {stderr:?}"
    );
}

/// Pids currently running in session `sid`, from a `/proc` scan. Used by the
/// session-kill tests to prove a session empties after cancellation.
fn session_pids(sid: libc::pid_t) -> Vec<libc::pid_t> {
    let mut pids = Vec::new();
    let Ok(entries) = std::fs::read_dir("/proc") else {
        return pids;
    };
    for entry in entries.flatten() {
        let name = entry.file_name();
        let Some(name) = name.to_str() else { continue };
        let Ok(pid) = name.parse::<libc::pid_t>() else {
            continue;
        };
        let Ok(stat) = std::fs::read_to_string(format!("/proc/{name}/stat")) else {
            continue;
        };
        let Some(rest) = stat.rsplit_once(')') else {
            continue;
        };
        let Some(rest) = rest.1.strip_prefix(' ') else {
            continue;
        };
        let mut f = rest.split(' ');
        let _ = f.next(); // state
        let _ = f.next(); // ppid
        let _ = f.next(); // pgrp
        if f.next().and_then(|s| s.parse::<libc::pid_t>().ok()) == Some(sid) {
            pids.push(pid);
        }
    }
    pids
}

/// Live (non-zombie) pids in the session — the emptiness definition used by
/// the Track B gate: a zombie is dead and only its reap timing is pending, so
/// it must not gate completion (init's reap pace is not under our control).
fn session_live_pids(sid: libc::pid_t) -> Vec<libc::pid_t> {
    let mut pids = Vec::new();
    let Ok(entries) = std::fs::read_dir("/proc") else {
        return pids;
    };
    for entry in entries.flatten() {
        let name = entry.file_name();
        let Some(name) = name.to_str() else { continue };
        let Ok(pid) = name.parse::<libc::pid_t>() else {
            continue;
        };
        let Ok(stat) = std::fs::read_to_string(format!("/proc/{name}/stat")) else {
            continue;
        };
        let Some(rest) = stat.rsplit_once(')') else {
            continue;
        };
        let Some(rest) = rest.1.strip_prefix(' ') else {
            continue;
        };
        let mut f = rest.split(' ');
        if f.next() == Some("Z") {
            continue;
        }
        let _ = f.next(); // ppid
        let _ = f.next(); // pgrp
        if f.next().and_then(|s| s.parse::<libc::pid_t>().ok()) == Some(sid) {
            pids.push(pid);
        }
    }
    pids
}

#[test]
fn parse_stat_starttime_extracts_field_22() {
    // "1 (systemd) S 0 1 1 0 -1 4194560 …" — starttime (field 22, index 19
    // after the comm-`)` split) is the identity token the sweep compares.
    let content = "123 (my proc) S 456 789 123 0 1 2 3 4 5 6 7 8 9 10 11 12 13 14 424242 15 16";
    // state..itrealvalue = 19 fields, then starttime = 424242.
    assert_eq!(crate::proc::parse_stat_starttime(content), Some(424242));
}

#[test]
fn parse_stat_starttime_tolerates_comm_with_spaces_and_parens() {
    // 19 fields after `) ` (state..itrealvalue), then starttime at index 19.
    let content = "7 (bash (with ) parens) R 1 7 7 0 -1 4194560 0 1 2 3 4 5 6 7 8 9 10 11 999 12";
    assert_eq!(crate::proc::parse_stat_starttime(content), Some(999));
}

#[test]
fn parse_stat_starttime_rejects_garbage() {
    assert_eq!(crate::proc::parse_stat_starttime("not a stat line"), None);
    assert_eq!(crate::proc::parse_stat_starttime("1 (x) R"), None);
}

/// Drive a [`ManagedProcess`] whose [`SpawnOptionsBuilder::chunk_sink`]
/// appends to `buffer`, calling `on_output` after every event batch with the
/// accumulated output and writing any returned bytes to the pty master. Used
/// by the interactive tests (Ctrl-C, job control) where input must be written
/// mid-stream once a marker line appears.
fn drive_managed_with_input(
    mut managed: crate::spawn::ManagedProcess,
    buffer: Arc<Mutex<Vec<u8>>>,
    mut on_output: impl FnMut(&str) -> Vec<u8>,
) -> Result<crate::spawn::Output, crate::CoreError> {
    let mut reactor = Reactor::new().unwrap();
    managed.register_with_reactor(&mut reactor).unwrap();
    let mut events = Vec::new();
    loop {
        if let Some(output) = managed.poll_completion(&mut reactor)? {
            return Ok(output);
        }
        let timeout = managed
            .next_deadline()
            .map(|at| {
                at.saturating_duration_since(std::time::Instant::now())
                    .as_millis()
                    .min(i32::MAX as u128) as i32
            })
            .unwrap_or(-1);
        reactor.wait(&mut events, 16, timeout)?;
        for event in &events {
            managed.handle_reactor_event(&mut reactor, event)?;
        }
        let text = String::from_utf8_lossy(&buffer.lock().unwrap()).to_string();
        let bytes = on_output(&text);
        if !bytes.is_empty() {
            managed.write_input(&bytes)?;
        }
    }
}

/// A sink that accumulates every chunk into `buffer` (never pauses).
fn accumulate_sink(buffer: Arc<Mutex<Vec<u8>>>) -> impl Fn(bool, &[u8]) -> SinkResult {
    move |_is_stdout, bytes| {
        buffer.lock().unwrap().extend_from_slice(bytes);
        SinkResult::Accept
    }
}

/// Standard pty options for the session-kill / job-control tests: isolated
/// group, session containment (the pty filter variant), streaming.
fn pty_contained_builder(
    argv: Vec<String>,
    backend: SpawnBackend,
    grace_ms: u32,
) -> SpawnOptionsBuilder {
    SpawnOptions::builder(argv, backend)
        .pgroup(ProcessGroup::new(None, true))
        .session_containment()
        .pty()
        .capture_stdout()
        .timeout_ms(15_000)
        .kill_grace_ms(grace_ms)
        .cancel(CancelPolicy::Graceful)
}

/// Standard pipe-mode options for the streamed spawns: isolated group (so the
/// child is a session leader), session containment, streaming. Mirrors the
/// daemon's pipe streaming path (`ProcessGroup::new(None, true)`).
fn pipe_contained_builder(
    argv: Vec<String>,
    backend: SpawnBackend,
    grace_ms: u32,
) -> SpawnOptionsBuilder {
    SpawnOptions::builder(argv, backend)
        .pgroup(ProcessGroup::new(None, true))
        .session_containment()
        .capture_stdout()
        .timeout_ms(15_000)
        .kill_grace_ms(grace_ms)
        .cancel(CancelPolicy::Graceful)
}

#[test]
fn test_pty_containment_allows_job_control_and_ctrl_c() {
    // The on-device defect, made a host test: under the strict filter
    // `setpgid` is denied, so a job never leaves the shell's pgrp, the tty's
    // foreground pgrp is the empty job group, and Ctrl-C kills nothing. Under
    // the pty containment variant `setpgid` is allowed, so a foreground job
    // lands in its own pgrp (== tty foreground pgrp) and SIGINT reaches it.
    for backend in [SpawnBackend::Fork, SpawnBackend::Clone3Pidfd] {
        let buffer = Arc::new(Mutex::new(Vec::new()));
        let sink_buf = Arc::clone(&buffer);
        let managed = spawn_managed(
            pty_contained_builder(
                vec![
                    "/bin/bash".to_string(),
                    "-c".to_string(),
                    "set -m; echo shellpgid=$(ps -o pgid= -p $$ | tr -d ' '); \
                     sh -c 'echo fgpgid=$(ps -o pgid= -p $$ | tr -d \" \"); sleep 30'"
                        .to_string(),
                ],
                backend,
                200,
            )
            .chunk_sink(accumulate_sink(sink_buf))
            .build()
            .unwrap(),
        )
        .unwrap();
        let sid = managed.pid();

        let ctrl_c_sent = Arc::new(AtomicBool::new(false));
        let sent = Arc::clone(&ctrl_c_sent);
        let output = drive_managed_with_input(managed, Arc::clone(&buffer), move |text| {
            if !sent.load(Ordering::SeqCst) && text.contains("fgpgid=") {
                sent.store(true, Ordering::SeqCst);
                vec![0x03] // Ctrl-C → SIGINT to the tty foreground pgrp
            } else {
                Vec::new()
            }
        })
        .unwrap();

        let merged = String::from_utf8_lossy(&buffer.lock().unwrap()).to_string();
        let shell_pgid = merged
            .lines()
            .find_map(|l| l.trim().strip_prefix("shellpgid=").map(str::trim))
            .expect("shellpgid marker");
        let fg_pgid = merged
            .lines()
            .find_map(|l| l.trim().strip_prefix("fgpgid=").map(str::trim))
            .expect("fgpgid marker");
        assert!(
            shell_pgid != fg_pgid && !shell_pgid.is_empty() && !fg_pgid.is_empty(),
            "{backend:?}: job control must put the foreground job in its own pgrp, got shell={shell_pgid} fg={fg_pgid}"
        );
        assert!(
            ctrl_c_sent.load(Ordering::SeqCst),
            "{backend:?}: Ctrl-C must have been sent (fg marker seen), got {merged:?}"
        );
        assert!(
            !output.timed_out,
            "{backend:?}: Ctrl-C must preempt the sleep before the timeout"
        );
        assert!(
            session_pids(sid).is_empty(),
            "{backend:?}: session must be empty after Ctrl-C, got {:?}",
            session_pids(sid)
        );
    }
}

#[test]
fn test_pty_session_kill_is_total_across_groups() {
    // A pty session fragmented into three pgrps (shell + two background jobs)
    // must all die on cancellation, and the drain must reach the master's EOF
    // — the "orphan wedges the drain" Horn-B scenario, closed.
    for backend in [SpawnBackend::Fork, SpawnBackend::Clone3Pidfd] {
        let managed = spawn_managed(
            pty_contained_builder(
                vec![
                    "/bin/bash".to_string(),
                    "-c".to_string(),
                    "set -m; sleep 60 & echo bg1=$!; sleep 60 & echo bg2=$!; wait".to_string(),
                ],
                backend,
                200,
            )
            .build()
            .unwrap(),
        )
        .unwrap();
        let sid = managed.pid();

        // Let bash print the bg markers and settle into `wait` before the
        // cancellation is requested.
        std::thread::sleep(std::time::Duration::from_millis(300));
        let started = std::time::Instant::now();
        let output = drive_managed(managed, true).unwrap();
        assert!(
            !output.timed_out,
            "{backend:?}: session kill must terminate the whole session, got {output:?}"
        );
        assert!(
            started.elapsed().as_secs() < 10,
            "{backend:?}: session kill must be prompt"
        );
        let merged = String::from_utf8_lossy(&output.stdout);
        let bg1: libc::pid_t = merged
            .lines()
            .find_map(|l| l.trim().strip_prefix("bg1="))
            .and_then(|s| s.trim().parse().ok())
            .expect("bg1 pid");
        let bg2: libc::pid_t = merged
            .lines()
            .find_map(|l| l.trim().strip_prefix("bg2="))
            .and_then(|s| s.trim().parse().ok())
            .expect("bg2 pid");
        assert!(
            session_pids(sid).is_empty(),
            "{backend:?}: session must be empty after cancellation, got {:?}",
            session_pids(sid)
        );
        assert_ne!(bg1, bg2);
    }
}

#[test]
fn test_pty_session_kill_continues_stopped_groups() {
    // A stopped background group must be SIGCONT'd before SIGTERM, otherwise
    // it would queue the TERM and stay immune. With a long grace period, the
    // job must die from the TERM phase (promptly), not from a SIGKILL
    // escalation — proving SIGCONT precedes the signal.
    let grace_ms = 8_000;
    let managed = spawn_managed(
        pty_contained_builder(
            vec![
                "/bin/bash".to_string(),
                "-c".to_string(),
                "set -m; sleep 60 & echo bg=$!; kill -STOP $!; wait".to_string(),
            ],
            SpawnBackend::Fork,
            grace_ms,
        )
        .build()
        .unwrap(),
    )
    .unwrap();
    let sid = managed.pid();

    let started = std::time::Instant::now();
    let output = drive_managed(managed, true).unwrap();
    let elapsed = started.elapsed().as_millis();
    assert!(
        !output.timed_out,
        "stopped group must be continued and die, got {output:?}"
    );
    assert!(
        elapsed < grace_ms as u128,
        "stopped group must die from TERM after SIGCONT, not the KILL escalation (elapsed {elapsed}ms vs grace {grace_ms}ms)"
    );
    assert!(session_pids(sid).is_empty(), "session must be empty");
}

#[test]
fn test_pty_session_kill_bounded_reenumeration_catches_raced_groups() {
    // A shell that keeps creating new background pgrps during cancellation
    // must still terminate: the kill loop re-enumerates while the master is
    // open, and the D-state give-up bound turns the re-enumeration into a
    // bounded chase rather than an unbounded /proc loop.
    let managed = spawn_managed(
        pty_contained_builder(
            vec![
                "/bin/bash".to_string(),
                    "-c".to_string(),
                    "set -m; i=0; while [ $i -lt 300 ]; do sleep 60 & i=$((i+1)); sleep 0.01; done; echo spawned=$i; wait".to_string(),
            ],
            SpawnBackend::Fork,
            200,
        )
        .build()
        .unwrap(),
    )
    .unwrap();
    let sid = managed.pid();

    let started = std::time::Instant::now();
    let output = drive_managed(managed, true).unwrap();
    assert!(
        started.elapsed().as_secs() < 12,
        "termination must stay bounded under a spawner, took {:?}",
        started.elapsed()
    );
    // The session may legitimately return with partial output (timed_out) if
    // the D-state give-up bound fired, but the session itself must empty once
    // the final SIGKILLs take effect.
    let mut wait = 0u32;
    while !session_pids(sid).is_empty() && wait < 50 {
        std::thread::sleep(std::time::Duration::from_millis(50));
        wait += 1;
    }
    assert!(
        session_pids(sid).is_empty(),
        "session must empty after bounded re-enumeration, still have {:?}",
        session_pids(sid)
    );
    let _ = output;
}

#[test]
fn test_pty_session_kill_ends_before_escalation_deadline() {
    // Master EOF before the TERM→KILL grace elapses: the session must finish
    // from the TERM phase, never reaching the SIGKILL escalation, so
    // `timed_out` stays false.
    let grace_ms = 8_000;
    let managed = spawn_managed(
        pty_contained_builder(
            vec![
                "/bin/bash".to_string(),
                "-c".to_string(),
                "set -m; sleep 30 & wait".to_string(),
            ],
            SpawnBackend::Fork,
            grace_ms,
        )
        .build()
        .unwrap(),
    )
    .unwrap();

    let started = std::time::Instant::now();
    let output = drive_managed(managed, true).unwrap();
    let elapsed = started.elapsed().as_millis();
    assert!(
        !output.timed_out,
        "clean TERM must EOF the master before the grace elapses, got {output:?}"
    );
    assert!(
        elapsed < grace_ms as u128,
        "must finish before the SIGKILL escalation (elapsed {elapsed}ms vs grace {grace_ms}ms)"
    );
}

#[test]
fn test_pty_session_kill_leader_only_leaves_background_alive() -> Result<(), crate::CoreError> {
    // Horn-F regression guard: killing only the session leader's pgrp — the
    // naive "signal the leader" route (§5b E1: `kill(-sid)` reaches only the
    // leader's group) — must leave a background job in its own pgrp alive and
    // the master open. Only the session-total enumeration (§5b E3) kills it.
    // Locks the E1/E3 distinction in so the kill can never be "simplified"
    // back to leader-only. F5 note: `LetMembersSurvive` keeps the natural
    // completion gate from sweeping the session (the F5 default `Sweep` would
    // SIGKILL the member on leader-reap), so the E1 window stays observable
    // and the E3 session-total kill stays the only path that empties it.
    let buffer = Arc::new(Mutex::new(Vec::new()));
    let sink_buf = Arc::clone(&buffer);
    let mut managed = spawn_managed(
        pty_contained_builder(
            vec![
                "/bin/bash".to_string(),
                "-c".to_string(),
                "set -m; sleep 60 & echo bg=$!; wait".to_string(),
            ],
            SpawnBackend::Fork,
            200,
        )
        .chunk_sink(accumulate_sink(sink_buf))
        .session_exit(SessionExitPolicy::LetMembersSurvive)
        .build()
        .unwrap(),
    )
    .unwrap();
    let sid = managed.pid();
    // setsid ⇒ the leader's pid is its session id AND its own pgrp.
    let leader_pgid = sid;

    let mut reactor = Reactor::new()?;
    managed.register_with_reactor(&mut reactor)?;
    let mut events = Vec::new();
    let mut bg_pid: Option<libc::pid_t> = None;
    let mut leader_killed = false;
    let started = std::time::Instant::now();
    loop {
        // E1 window: observe the session via /proc WITHOUT driving
        // poll_completion — under `LetMembersSurvive` completion fires on
        // leader-reap, so polling it here would consume the session before
        // the E1 assertion. The reactor still runs so output drains.
        if !leader_killed {
            let text = String::from_utf8_lossy(&buffer.lock().unwrap()).to_string();
            if let Some(rest) = text.lines().find_map(|l| l.trim().strip_prefix("bg=")) {
                let bg = rest.trim().parse::<libc::pid_t>().ok();
                if bg.is_some_and(|bg| session_pids(sid).contains(&bg)) {
                    bg_pid = bg;
                    // Leader-only route (Horn F): signal just the leader's group.
                    unsafe {
                        libc::kill(-leader_pgid, libc::SIGTERM);
                    }
                    leader_killed = true;
                }
            }
        } else if let Some(bg) = bg_pid
            && started.elapsed() > std::time::Duration::from_millis(1500)
        {
            assert!(
                session_pids(sid).contains(&bg),
                "leader-only kill must leave the background job alive (bg={bg}, session={:?})",
                session_pids(sid)
            );
            managed.request_cancel();
            let output = drive_managed(managed, false)?;
            assert!(
                !output.timed_out,
                "session-total kill must terminate the background job, got {output:?}"
            );
            assert!(
                session_pids(sid).is_empty(),
                "session must be empty after session-total kill, got {:?}",
                session_pids(sid)
            );
            return Ok(());
        }
        reactor.wait(&mut events, 16, 50)?;
        for event in &events {
            managed.handle_reactor_event(&mut reactor, event)?;
        }
    }
}

#[test]
fn test_pty_session_enumeration_only_on_termination() -> Result<(), crate::CoreError> {
    // The /proc session enumeration is a per-termination control-plane
    // operation, never a data-path one: a live session doing output, input and
    // resize performs zero walks. Only cancellation/drop enumerates. Tracked
    // per-session so parallel pty tests don't interfere.
    reset_pty_enumeration();
    let buffer = Arc::new(Mutex::new(Vec::new()));
    let sink_buf = Arc::clone(&buffer);
    let mut managed = spawn_managed(
        pty_contained_builder(vec!["/bin/cat".to_string()], SpawnBackend::Fork, 200)
            .chunk_sink(accumulate_sink(sink_buf))
            .build()
            .unwrap(),
    )
    .unwrap();
    let sid = managed.pid();

    let mut reactor = Reactor::new()?;
    managed.register_with_reactor(&mut reactor)?;
    let mut events = Vec::new();
    let started = std::time::Instant::now();
    let mut wrote = false;
    let mut resized = false;
    let mut cancelled = false;
    loop {
        if let Some(_output) = managed.poll_completion(&mut reactor)? {
            panic!("cat must stay alive until cancelled");
        }
        let timeout = managed
            .next_deadline()
            .map(|at| {
                at.saturating_duration_since(std::time::Instant::now())
                    .as_millis()
                    .min(i32::MAX as u128) as i32
            })
            .unwrap_or(-1);
        reactor.wait(&mut events, 16, timeout)?;
        for event in &events {
            managed.handle_reactor_event(&mut reactor, event)?;
        }
        let elapsed = started.elapsed();

        if !wrote {
            managed.write_input(b"hello enumeration\n")?;
            wrote = true;
        }
        if !resized && elapsed > std::time::Duration::from_millis(120) {
            managed.resize_pty(50, 200)?;
            resized = true;
        }
        assert!(
            !pty_session_enumerated(sid),
            "streaming/output/write/resize must never enumerate /proc"
        );
        if !cancelled && resized && elapsed > std::time::Duration::from_millis(250) {
            managed.request_cancel();
            cancelled = true;
        }
        if cancelled && elapsed > std::time::Duration::from_millis(250) {
            break;
        }
    }
    assert!(
        !pty_session_enumerated(sid),
        "no /proc enumeration before termination"
    );

    let output = drive_managed(managed, false)?;
    let _ = output;
    assert!(
        pty_session_enumerated(sid),
        "termination must enumerate the session's PGIDs"
    );
    assert!(session_pids(sid).is_empty(), "session must be empty");
    Ok(())
}

#[test]
fn test_pty_natural_completion_waits_for_detached_background() -> Result<(), crate::CoreError> {
    // H1 (pty): master EOF + leader reap must NOT report COMPLETED while a
    // slave-detached background member still runs. The shell exits cleanly and
    // `sleep` inherits no slave fd, so the master EOFs immediately; only the
    // session-emptiness gate (Track B) sweeps the contained member and gates
    // completion until /proc shows the session empty.
    let buffer = Arc::new(Mutex::new(Vec::new()));
    let sink_buf = Arc::clone(&buffer);
    let mut managed = spawn_managed(
        pty_contained_builder(
            vec![
                "/bin/sh".to_string(),
                "-c".to_string(),
                "trap '' HUP; sleep 1000 </dev/null >/dev/null 2>&1 & echo bg=$!; exit 0"
                    .to_string(),
            ],
            SpawnBackend::Fork,
            200,
        )
        .chunk_sink(accumulate_sink(sink_buf))
        .build()
        .unwrap(),
    )
    .unwrap();
    let sid = managed.pid();
    let mut reactor = Reactor::new()?;
    managed.register_with_reactor(&mut reactor)?;
    let mut events = Vec::new();
    let mut bg_pid: Option<libc::pid_t> = None;
    let mut bg_seen_live = false;
    let mut output = None;
    let started = std::time::Instant::now();
    loop {
        // Observe the session BEFORE `poll_completion`: the sweep fires in the
        // same call that reaps the leader, so the "detached member alive after
        // leader exit" state is only visible at loop-top (the leader is still a
        // zombie in /proc until `wait_step` reaps it).
        let text = String::from_utf8_lossy(&buffer.lock().unwrap()).to_string();
        if bg_pid.is_none()
            && let Some(rest) = text.lines().find_map(|l| l.trim().strip_prefix("bg="))
        {
            bg_pid = rest.trim().parse::<libc::pid_t>().ok();
        }
        if let Some(bg) = bg_pid
            && session_pids(sid).contains(&bg)
            && unsafe { libc::kill(bg, 0) } == 0
        {
            bg_seen_live = true;
        }
        if let Some(out) = managed.poll_completion(&mut reactor)? {
            output = Some(out);
            break;
        }
        let timeout = managed
            .next_deadline()
            .map(|at| {
                at.saturating_duration_since(std::time::Instant::now())
                    .as_millis()
                    .min(i32::MAX as u128) as i32
            })
            .unwrap_or(-1);
        reactor.wait(&mut events, 16, timeout)?;
        for event in &events {
            managed.handle_reactor_event(&mut reactor, event)?;
        }
        if started.elapsed() > std::time::Duration::from_secs(10) {
            break;
        }
    }
    let output = output.expect("natural completion must eventually be reported");
    assert!(
        bg_seen_live,
        "test setup broken: the detached background member was never observed live in the session"
    );
    assert!(
        !output.timed_out,
        "natural completion must not time out, got {output:?}"
    );
    assert!(
        matches!(output.status, Some(crate::spawn::ExitStatus::Exited(0))),
        "shell must exit 0, got {:?}",
        output.status
    );
    assert!(
        session_live_pids(sid).is_empty(),
        "completion must be gated on an empty session, got {:?}",
        session_live_pids(sid)
    );
    if let Some(bg) = bg_pid {
        assert!(
            unsafe { libc::kill(bg, 0) } != 0,
            "detached background member must be swept before completion (bg={bg})"
        );
    }
    Ok(())
}

#[test]
fn test_pty_natural_completion_sweeps_slave_holding_background() -> Result<(), crate::CoreError> {
    // F5 (A4-3): the natural pty gate `(io_done || paused) && sweep_if_leader()`
    // hangs forever when a background member HOLDS the slave — the master never
    // EOFs, so `io_done` stays false and the sweep never runs. Completion must
    // instead be gated on leader-reap + session sweep (with a sweep bound), so a
    // slave-holding member cannot stall the job indefinitely.
    let buffer = Arc::new(Mutex::new(Vec::new()));
    let sink_buf = Arc::clone(&buffer);
    let mut managed = spawn_managed(
        pty_contained_builder(
            vec![
                "/bin/sh".to_string(),
                "-c".to_string(),
                // NO stdio redirect: the bg member inherits the pty slave, so the
                // master stays open after the shell exits (the A4-3 hang).
                "trap '' HUP; sleep 1000 & echo bg=$!; exit 0".to_string(),
            ],
            SpawnBackend::Fork,
            200,
        )
        .chunk_sink(accumulate_sink(sink_buf))
        .build()
        .unwrap(),
    )
    .unwrap();
    let sid = managed.pid();
    let mut reactor = Reactor::new()?;
    managed.register_with_reactor(&mut reactor)?;
    let mut events = Vec::new();
    let mut bg_pid: Option<libc::pid_t> = None;
    let mut bg_seen_live = false;
    let mut output = None;
    let started = std::time::Instant::now();
    loop {
        let text = String::from_utf8_lossy(&buffer.lock().unwrap()).to_string();
        if bg_pid.is_none()
            && let Some(rest) = text.lines().find_map(|l| l.trim().strip_prefix("bg="))
        {
            bg_pid = rest.trim().parse::<libc::pid_t>().ok();
        }
        if let Some(bg) = bg_pid
            && session_pids(sid).contains(&bg)
            && unsafe { libc::kill(bg, 0) } == 0
        {
            bg_seen_live = true;
        }
        if let Some(out) = managed.poll_completion(&mut reactor)? {
            output = Some(out);
            break;
        }
        let timeout = managed
            .next_deadline()
            .map(|at| {
                at.saturating_duration_since(std::time::Instant::now())
                    .as_millis()
                    .min(i32::MAX as u128) as i32
            })
            .unwrap_or(-1);
        reactor.wait(&mut events, 16, timeout)?;
        for event in &events {
            managed.handle_reactor_event(&mut reactor, event)?;
        }
        if started.elapsed() > std::time::Duration::from_secs(10) {
            break;
        }
    }
    let output = output.expect("natural completion must not hang on a slave-holding member");
    assert!(
        bg_seen_live,
        "test setup broken: the slave-holding background member was never observed live in the session"
    );
    assert!(
        matches!(output.status, Some(crate::spawn::ExitStatus::Exited(0))),
        "shell must exit 0, got {:?}",
        output.status
    );
    assert!(
        session_live_pids(sid).is_empty(),
        "completion must be gated on an empty session, got {:?}",
        session_live_pids(sid)
    );
    assert!(
        output.swept_members,
        "F14: the natural-completion sweep that emptied the session must be \
         surfaced on Output (slave-holding member was SIGKILLed), got {:?}",
        output.swept_members
    );
    if let Some(bg) = bg_pid {
        assert!(
            unsafe { libc::kill(bg, 0) } != 0,
            "slave-holding background member must be swept before completion (bg={bg})"
        );
    }
    Ok(())
}

#[test]
fn test_let_members_survive_reports_no_swept_members() -> Result<(), crate::CoreError> {
    // F14: under `SessionExitPolicy::LetMembersSurvive` the natural-completion
    // sweep never runs, so `Output::swept_members` must be false even when a
    // contained member outlives the leader. Completion is still gated on
    // leader-reap (no EOF hang), but the session is not signaled.
    let buffer = Arc::new(Mutex::new(Vec::new()));
    let sink_buf = Arc::clone(&buffer);
    let mut managed = spawn_managed(
        pty_contained_builder(
            vec![
                "/bin/sh".to_string(),
                "-c".to_string(),
                // NO stdio redirect: the bg member inherits the pty slave.
                "trap '' HUP; sleep 1000 & echo bg=$!; exit 0".to_string(),
            ],
            SpawnBackend::Fork,
            200,
        )
        .chunk_sink(accumulate_sink(sink_buf))
        .session_exit(SessionExitPolicy::LetMembersSurvive)
        .build()
        .unwrap(),
    )
    .unwrap();
    let sid = managed.pid();
    let mut reactor = Reactor::new()?;
    managed.register_with_reactor(&mut reactor)?;
    let mut events = Vec::new();
    let mut output = None;
    let started = std::time::Instant::now();
    loop {
        if let Some(out) = managed.poll_completion(&mut reactor)? {
            output = Some(out);
            break;
        }
        let timeout = managed
            .next_deadline()
            .map(|at| {
                at.saturating_duration_since(std::time::Instant::now())
                    .as_millis()
                    .min(i32::MAX as u128) as i32
            })
            .unwrap_or(-1);
        reactor.wait(&mut events, 16, timeout)?;
        for event in &events {
            managed.handle_reactor_event(&mut reactor, event)?;
        }
        if started.elapsed() > std::time::Duration::from_secs(10) {
            break;
        }
    }
    let output = output.expect("LetMembersSurvive must complete on leader-reap");
    assert!(
        !output.swept_members,
        "F14: LetMembersSurvive must not report swept members, got {:?}",
        output.swept_members
    );
    assert!(
        matches!(output.status, Some(crate::spawn::ExitStatus::Exited(0))),
        "shell must exit 0, got {:?}",
        output.status
    );
    // The surviving member keeps the slave: completion force-closes the
    // master. The contained member is NOT swept (kill-totality opt-out).
    assert!(
        !session_live_pids(sid).is_empty(),
        "LetMembersSurvive must leave contained members running"
    );
    // Cleanup: Drop cannot sweep a completed process (`running` was already
    // consumed by finish) and must not — it would contradict the
    // LetMembersSurvive opt-out. The test reaps the surviving member
    // directly so no root child leaks.
    let text = String::from_utf8_lossy(&buffer.lock().unwrap()).to_string();
    let bg = text
        .lines()
        .find_map(|l| l.trim().strip_prefix("bg="))
        .and_then(|rest| rest.trim().parse::<libc::pid_t>().ok())
        .expect("test setup broken: bg= never printed");
    assert_eq!(unsafe { libc::kill(bg, libc::SIGKILL) }, 0, "cleanup kill");
    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
    while !session_live_pids(sid).is_empty() {
        assert!(
            std::time::Instant::now() < deadline,
            "direct SIGKILL must empty the session, sid={sid}"
        );
        std::thread::sleep(std::time::Duration::from_millis(100));
    }
    Ok(())
}

#[test]
fn test_pipe_natural_completion_waits_for_detached_background() -> Result<(), crate::CoreError> {
    // H6 (pipe, streamed): EOF + leader reap must NOT report COMPLETED while
    // an fd-detached background member still runs. `sleep` redirects its
    // stdio to /dev/null, so the pipes EOF as the shell exits; only the
    // session-emptiness gate (Track B) sweeps it before completion.
    let buffer = Arc::new(Mutex::new(Vec::new()));
    let sink_buf = Arc::clone(&buffer);
    let mut managed = spawn_managed(
        pipe_contained_builder(
            vec![
                "/bin/sh".to_string(),
                "-c".to_string(),
                "trap '' HUP; sleep 1000 </dev/null >/dev/null 2>&1 & echo bg=$!; exit 0"
                    .to_string(),
            ],
            SpawnBackend::Fork,
            200,
        )
        .chunk_sink(accumulate_sink(sink_buf))
        .build()
        .unwrap(),
    )
    .unwrap();
    let sid = managed.pid();
    let mut reactor = Reactor::new()?;
    managed.register_with_reactor(&mut reactor)?;
    let mut events = Vec::new();
    let mut bg_pid: Option<libc::pid_t> = None;
    let mut bg_seen_live = false;
    let mut output = None;
    let started = std::time::Instant::now();
    loop {
        // Observe the session BEFORE `poll_completion`: the sweep fires in the
        // same call that reaps the leader, so the "detached member alive after
        // leader exit" state is only visible at loop-top (the leader is still a
        // zombie in /proc until `wait_step` reaps it).
        let text = String::from_utf8_lossy(&buffer.lock().unwrap()).to_string();
        if bg_pid.is_none()
            && let Some(rest) = text.lines().find_map(|l| l.trim().strip_prefix("bg="))
        {
            bg_pid = rest.trim().parse::<libc::pid_t>().ok();
        }
        if let Some(bg) = bg_pid
            && session_pids(sid).contains(&bg)
            && unsafe { libc::kill(bg, 0) } == 0
        {
            bg_seen_live = true;
        }
        if let Some(out) = managed.poll_completion(&mut reactor)? {
            output = Some(out);
            break;
        }
        let timeout = managed
            .next_deadline()
            .map(|at| {
                at.saturating_duration_since(std::time::Instant::now())
                    .as_millis()
                    .min(i32::MAX as u128) as i32
            })
            .unwrap_or(-1);
        reactor.wait(&mut events, 16, timeout)?;
        for event in &events {
            managed.handle_reactor_event(&mut reactor, event)?;
        }
        if started.elapsed() > std::time::Duration::from_secs(10) {
            break;
        }
    }
    let output = output.expect("natural completion must eventually be reported");
    assert!(
        bg_seen_live,
        "test setup broken: the detached background member was never observed live in the session"
    );
    assert!(
        !output.timed_out,
        "natural completion must not time out, got {output:?}"
    );
    assert!(
        matches!(output.status, Some(crate::spawn::ExitStatus::Exited(0))),
        "shell must exit 0, got {:?}",
        output.status
    );
    assert!(
        session_live_pids(sid).is_empty(),
        "completion must be gated on an empty session, got {:?}",
        session_live_pids(sid)
    );
    if let Some(bg) = bg_pid {
        assert!(
            unsafe { libc::kill(bg, 0) } != 0,
            "detached background member must be swept before completion (bg={bg})"
        );
    }
    Ok(())
}

#[test]
fn test_spawn_blocking_natural_completion_sweeps_detached_background() {
    // H6 (pipe, blocking `spawn`/`run`): the wait_loop gate must sweep the
    // fd-detached background member before the blocking call returns, so the
    // returned session is empty and the job is never reported exit-0 while a
    // contained member still runs.
    let out = SpawnOptions::builder(
        vec![
            "/bin/sh".to_string(),
            "-c".to_string(),
            "trap '' HUP; (sleep 1000 </dev/null >/dev/null 2>&1 &); exit 0".to_string(),
        ],
        SpawnBackend::Fork,
    )
    .pgroup(ProcessGroup::new(None, true))
    .session_containment()
    .capture_stdout()
    .timeout_ms(10_000)
    .kill_grace_ms(200)
    .cancel(CancelPolicy::Graceful)
    .build()
    .unwrap()
    .run()
    .unwrap();
    let sid = out.pid;
    assert!(
        matches!(out.status, Some(crate::spawn::ExitStatus::Exited(0))),
        "shell must exit 0, got {:?}",
        out.status
    );
    assert!(
        !out.timed_out,
        "blocking spawn must not time out, got {out:?}"
    );
    assert!(
        session_live_pids(sid).is_empty(),
        "completion must be gated on an empty session, got {:?}",
        session_live_pids(sid)
    );
    assert!(
        out.swept_members,
        "F14: the natural-completion sweep that emptied the session must be \
         surfaced on Output (detached member was SIGKILLed), got {:?}",
        out.swept_members
    );
}

#[test]
fn test_pipe_cancel_sweeps_leader_reap_survivor() -> Result<(), crate::CoreError> {
    // H4 (pipe, streamed): the non-pty cancel kill stops once the leader is
    // reaped, so a TERM-immune background member that survives the group TERM
    // would escape unmanaged. The H4 completion gate must keep sweeping the
    // contained session until /proc shows no live members before the cancel
    // reports back. The background `sleep` ignores TERM (inherited from
    // `trap '' TERM`) and holds the stdout pipe open, so the drain cannot
    // close on its own; the large kill grace ensures the group-SIGKILL
    // escalation can never fire within the test window, leaving the H4 sweep
    // as the only possible killer of the survivor.
    let buffer = Arc::new(Mutex::new(Vec::new()));
    let sink_buf = Arc::clone(&buffer);
    let mut managed = spawn_managed(
        pipe_contained_builder(
            vec![
                "/bin/sh".to_string(),
                "-c".to_string(),
                "trap '' TERM; sleep 1000 & echo bg=$!; trap - TERM; sleep 30".to_string(),
            ],
            SpawnBackend::Fork,
            3_000,
        )
        .timeout_ms(1_000)
        .chunk_sink(accumulate_sink(sink_buf))
        .build()
        .unwrap(),
    )
    .unwrap();
    let sid = managed.pid();
    let mut reactor = Reactor::new()?;
    managed.register_with_reactor(&mut reactor)?;
    let mut events = Vec::new();
    let mut bg_pid: Option<libc::pid_t> = None;
    let mut bg_seen_live = false;
    let mut output = None;
    let started = std::time::Instant::now();
    loop {
        // Observe the session BEFORE `poll_completion`: once the deadline
        // fires, the leader is TERMed and the H4 sweep runs on the same call
        // that reaps it. The fixed 50 ms cadence (rather than
        // `next_deadline`, which would sleep until the grace escalation) is
        // what reaps the leader before the sweep has to kill its survivor.
        let text = String::from_utf8_lossy(&buffer.lock().unwrap()).to_string();
        if bg_pid.is_none()
            && let Some(rest) = text.lines().find_map(|l| l.trim().strip_prefix("bg="))
        {
            bg_pid = rest.trim().parse::<libc::pid_t>().ok();
        }
        if let Some(bg) = bg_pid
            && session_pids(sid).contains(&bg)
            && unsafe { libc::kill(bg, 0) } == 0
        {
            bg_seen_live = true;
        }
        if let Some(out) = managed.poll_completion(&mut reactor)? {
            output = Some(out);
            break;
        }
        reactor.wait(&mut events, 16, 50)?;
        for event in &events {
            managed.handle_reactor_event(&mut reactor, event)?;
        }
        if started.elapsed() > std::time::Duration::from_secs(10) {
            break;
        }
    }
    let output = output.expect("cancel must eventually be reported");
    assert!(
        bg_seen_live,
        "test setup broken: the TERM-immune background member was never observed live in the session"
    );
    assert!(
        output.timed_out,
        "cancel must report timed_out, got {output:?}"
    );
    assert!(
        output.status.is_some(),
        "the leader must be reaped on cancel, got {:?}",
        output.status
    );
    assert!(
        session_live_pids(sid).is_empty(),
        "cancel completion must be gated on an empty session, got {:?}",
        session_live_pids(sid)
    );
    if let Some(bg) = bg_pid {
        assert!(
            unsafe { libc::kill(bg, 0) } != 0,
            "TERM-immune background member must be swept before cancel completion (bg={bg})"
        );
    }
    Ok(())
}

#[test]
fn test_spawn_blocking_cancel_sweeps_leader_reap_survivor() {
    // H4 (pipe, blocking `spawn`/`run`): the wait_loop timed-out branch must
    // sweep the TERM-immune member that holds the pipe write end after the
    // leader is reaped (the group kill stops at leader reap) — the blocking
    // call returns timed-out only once the contained session is empty. The
    // large kill grace keeps the group-SIGKILL escalation from firing within
    // the test window, so the sweep is the only possible killer.
    let out = SpawnOptions::builder(
        vec![
            "/bin/sh".to_string(),
            "-c".to_string(),
            "trap '' TERM; sleep 1000 & echo bg=$!; trap - TERM; sleep 30".to_string(),
        ],
        SpawnBackend::Fork,
    )
    .pgroup(ProcessGroup::new(None, true))
    .session_containment()
    .capture_stdout()
    .timeout_ms(1_000)
    .kill_grace_ms(3_000)
    .cancel(CancelPolicy::Graceful)
    .build()
    .unwrap()
    .run()
    .unwrap();
    let sid = out.pid;
    let merged = String::from_utf8_lossy(&out.stdout);
    let bg = merged
        .lines()
        .find_map(|l| l.trim().strip_prefix("bg=").map(str::trim))
        .and_then(|s| s.parse::<libc::pid_t>().ok())
        .expect("bg= marker must be present in the captured output");
    assert!(
        out.timed_out,
        "blocking cancel must report timed_out, got {out:?}"
    );
    assert!(
        out.status.is_some(),
        "the leader must be reaped on cancel, got {:?}",
        out.status
    );
    assert!(
        session_live_pids(sid).is_empty(),
        "blocking cancel must sweep the contained session before returning, got {:?}",
        session_live_pids(sid)
    );
    assert!(
        unsafe { libc::kill(bg, 0) } != 0,
        "TERM-immune background member must be swept before the blocking call returns (bg={bg})"
    );
}

#[test]
fn test_orphan_session_reaper_sweeps_detached_members() -> Result<(), crate::CoreError> {
    // H2 (white-box, mechanism test): the D-state give-up fires when the
    // session *leader* is unreapable — it cannot keep polling forever, but the
    // kill must not stop either, or a member that survived the escalation
    // (D-state, or forked into the final sweep window) leaks unmanaged. The
    // give-up hands the session to the detached reaper, which keeps SIGKILLing
    // every group until /proc shows no live members. Here we register a live,
    // running session directly (the leader is still alive, so the sid is pinned
    // exactly as at the D-state give-up) and assert the reaper converges on
    // both the TERM-immune background member and the leader itself. End-to-end
    // reproduction of the give-up would require a genuinely D-state (kernel
    // uninterruptible-I/O) leader, which cannot be produced deterministically
    // from userspace; the white-box call is the reachable half of the contract.
    let buffer = Arc::new(Mutex::new(Vec::new()));
    let sink_buf = Arc::clone(&buffer);
    let mut managed = spawn_managed(
        pipe_contained_builder(
            vec![
                "/bin/sh".to_string(),
                "-c".to_string(),
                "trap '' TERM; sleep 1000 & echo bg=$!; sleep 30".to_string(),
            ],
            SpawnBackend::Fork,
            200,
        )
        .chunk_sink(accumulate_sink(sink_buf))
        .build()
        .unwrap(),
    )
    .unwrap();
    let sid = managed.pid();
    let mut reactor = Reactor::new()?;
    managed.register_with_reactor(&mut reactor)?;
    let mut events = Vec::new();
    let mut bg_pid: Option<libc::pid_t> = None;
    let started = std::time::Instant::now();
    while bg_pid.is_none() {
        let text = String::from_utf8_lossy(&buffer.lock().unwrap()).to_string();
        if let Some(rest) = text.lines().find_map(|l| l.trim().strip_prefix("bg=")) {
            let bg = rest.trim().parse::<libc::pid_t>().ok();
            if bg.is_some_and(|bg| session_pids(sid).contains(&bg)) {
                bg_pid = bg;
                break;
            }
        }
        let _ = managed.poll_completion(&mut reactor)?;
        reactor.wait(&mut events, 16, 20)?;
        for event in &events {
            managed.handle_reactor_event(&mut reactor, event)?;
        }
        assert!(
            started.elapsed() < std::time::Duration::from_secs(5),
            "the TERM-immune background member was never observed live in the session"
        );
    }
    let bg = bg_pid.unwrap();
    // Hand the live session to the detached reaper, exactly as the D-state
    // give-up sites do.
    orphan_session(sid);
    let mut emptied = false;
    for _ in 0..40 {
        if session_live_pids(sid).is_empty() && unsafe { libc::kill(bg, 0) } != 0 {
            emptied = true;
            break;
        }
        std::thread::sleep(std::time::Duration::from_millis(100));
    }
    assert!(
        emptied,
        "detached reaper must empty the orphaned session, live={:?} bg={bg}",
        session_live_pids(sid)
    );
    Ok(())
}

#[test]
fn reaper_prune_removes_only_confirmed_reaped_pids() {
    // F3/F4 (race regression): the old reaper built `still_orphaned` from a
    // snapshot and `retain`ed the set to it — any pid registered between the
    // snapshot and the retain was silently evicted (never reaped, never
    // re-swept). The fix removes only the pids this pass *confirmed* reaped,
    // so a concurrent registration survives. Drive the pure seam directly:
    // set = {a, b}, snapshot saw only {a}, a is confirmed reaped.
    let mut set = std::collections::HashSet::new();
    set.insert(101);
    set.insert(202);
    let snapshot: Vec<libc::pid_t> = vec![101];
    let reaped: Vec<libc::pid_t> = snapshot
        .iter()
        .copied()
        .filter(|pid| *pid == 101) // waitpid would confirm 101 reaped
        .collect();
    crate::spawn::prune_reaped_by_reaped_only(&mut set, &reaped);
    assert!(!set.contains(&101), "reaped pid must be pruned");
    assert!(
        set.contains(&202),
        "pid registered concurrently must survive the prune"
    );
}

#[test]
fn orphan_session_refuses_unreadable_or_recycled_sid() {
    // rev6-F1: an orphaned-session registration whose leader is already gone
    // (or whose starttime cannot be verified) must be refused — sweeping by a
    // bare numeric sid after the leader is reaped could SIGKILL an unrelated
    // recycled session. A pid that never existed has no starttime.
    let nonexistent = 1_000_000_000;
    assert_eq!(crate::proc::starttime(nonexistent), None);
    crate::spawn::orphan_session(nonexistent);
    assert_eq!(
        crate::spawn::orphaned_sessions_len(),
        0,
        "a sid whose leader starttime is unreadable must not be registered"
    );
}

#[test]
fn session_scan_failures_are_fail_closed() {
    // F7: a `/proc` scan that cannot be read must surface as `Err`, never as a
    // fabricated empty session. Before F7, `session_pgids` returned an empty
    // set and `session_sweep` returned `true` on `read_dir` failure — the job
    // reported COMPLETED with contained members still live (fail-open). A
    // nonexistent proc root is the deterministic reproduction.
    let bad_root = std::path::Path::new("/nonexistent");
    assert!(
        crate::spawn::session_pgids_at(bad_root, 1234).is_err(),
        "unreadable /proc must surface as Err, not an empty set"
    );
    assert!(
        crate::spawn::session_sweep_at(bad_root, 1234).is_err(),
        "unreadable /proc must surface as Err, not a false 'session empty'"
    );
}

#[test]
fn session_scan_seam_reads_a_real_root() {
    // The `_at` seams must agree with the `/proc`-rooted production path on a
    // readable root — otherwise the seam is testing a different function. The
    // sweep path kills live members, so both checks run against an isolated
    // child in its own session, never the caller's.
    let pid = unsafe { libc::fork() };
    assert!(pid >= 0);
    if pid == 0 {
        // Child: new session (sid == its own pid), live forever until killed.
        unsafe {
            libc::setsid();
            libc::pause();
        }
        unsafe { libc::_exit(0) };
    }
    let sid = pid; // setsid() makes the child its own session leader
    // Enumeration sees the child in its own session.
    let pgids = crate::spawn::session_pgids_at(std::path::Path::new("/proc"), sid).unwrap();
    assert!(
        pgids.contains(&sid),
        "the child's own pgid must be enumerable in its own session, got {pgids:?}"
    );
    // The child is live in its session: sweep returns Ok(false) (non-empty)
    // and SIGKILLs it.
    assert!(
        !crate::spawn::session_sweep_at(std::path::Path::new("/proc"), sid).unwrap(),
        "a session with a live member must sweep as non-empty"
    );
    let mut status = 0;
    loop {
        let r = unsafe { libc::waitpid(sid, &mut status, 0) };
        if r == sid {
            break;
        }
        assert_eq!(r, -1, "waitpid of swept child failed");
        assert_eq!(
            std::io::Error::last_os_error().raw_os_error(),
            Some(libc::EINTR)
        );
    }
    // Once the member is gone (reaped), the same sid sweeps as empty.
    assert!(
        crate::spawn::session_sweep_at(std::path::Path::new("/proc"), sid).unwrap(),
        "a session with no live member must sweep as empty"
    );
}

#[test]
fn drop_pty_managed_process_kills_session_members() -> Result<(), crate::CoreError> {
    // F2/RVB-1 (Drop re-spec): when the session *leader* has already been
    // reaped by `poll_completion` while a TERM-immune background member still
    // holds the pty slave, the old `status.is_some()` early-return in Drop
    // skipped every signal — the member leaked. The early-return is the
    // recycled-pid guard and must stay for non-session spawns, but a session
    // spawn must sweep (freshness check) even with a reaped leader. Construct
    // exactly that: the shell exits 0 (leader reaped, `status.is_some()`),
    // the bg `sleep` survives holding the slave (no stdio redirects), then
    // Drop runs (daemon `shutdown_all` path) and must empty the session.
    let buffer = Arc::new(Mutex::new(Vec::new()));
    let sink_buf = Arc::clone(&buffer);
    let mut managed = spawn_managed(
        pipe_contained_builder(
            vec![
                "/bin/sh".to_string(),
                "-c".to_string(),
                "trap '' HUP; sleep 1000 & echo bg=$!; exit 0".to_string(),
            ],
            SpawnBackend::Fork,
            200,
        )
        .pty()
        .chunk_sink(accumulate_sink(sink_buf))
        .build()
        .unwrap(),
    )
    .unwrap();
    let sid = managed.pid();
    let mut reactor = Reactor::new()?;
    managed.register_with_reactor(&mut reactor)?;
    let mut events = Vec::new();
    let mut bg_pid: Option<libc::pid_t> = None;
    let started = std::time::Instant::now();
    // Wait until the bg member is live AND the leader has been reaped by
    // poll_completion (status.is_some()) — the natural gate stays open because
    // the member holds the slave open (io_done false).
    loop {
        let text = String::from_utf8_lossy(&buffer.lock().unwrap()).to_string();
        if bg_pid.is_none()
            && let Some(rest) = text.lines().find_map(|l| l.trim().strip_prefix("bg="))
        {
            let bg = rest.trim().parse::<libc::pid_t>().ok();
            if bg.is_some_and(|bg| session_pids(sid).contains(&bg)) {
                bg_pid = bg;
            }
        }
        if let Some(output) = managed.poll_completion(&mut reactor)? {
            // F5: the natural gate completes only once the sweep has emptied
            // the session (Sweep default). If it completed, the member is
            // already dead — the Drop path is not the one under test here,
            // but the empty-session invariant must hold.
            let live = session_live_pids(sid);
            assert!(
                live.is_empty(),
                "natural completion under Sweep must empty the session: \
                 output={output:?} live={live:?}"
            );
            return Ok(());
        }
        if bg_pid.is_some()
            && !std::path::Path::new(&format!("/proc/{sid}")).exists()
            && !session_pids(sid).is_empty()
        {
            break;
        }
        reactor.wait(&mut events, 16, 20)?;
        for event in &events {
            managed.handle_reactor_event(&mut reactor, event)?;
        }
        assert!(
            started.elapsed() < std::time::Duration::from_secs(5),
            "leader never reaped with a live member (bg={bg_pid:?} sid={sid})"
        );
    }
    let bg = bg_pid.unwrap();
    // Drop the managed process with the leader already reaped. The member must
    // be killed and the session emptied — Drop is the containment backstop.
    drop(managed);
    let mut emptied = false;
    for _ in 0..40 {
        if session_live_pids(sid).is_empty() && unsafe { libc::kill(bg, 0) } != 0 {
            emptied = true;
            break;
        }
        std::thread::sleep(std::time::Duration::from_millis(100));
    }
    assert!(
        emptied,
        "Drop must empty the session even with a reaped leader, live={:?} bg={bg}",
        session_live_pids(sid)
    );
    Ok(())
}

#[test]
fn test_pipe_containment_still_denies_setpgid() {
    // Pipe mode keeps the strict filter: a job-control shell's setpgid for a
    // background job must fail, leaving the job in the shell's pgrp (the same
    // directory the on-device defect lived in, but for pipe mode — unchanged).
    let out = SpawnOptions::builder(
        vec![
            "/bin/bash".to_string(),
            "-c".to_string(),
            "set -m; sleep 30 & echo bg=$!; sleep 0.3; \
             echo shell=$(ps -o pgid= -p $$ | tr -d ' '); \
             echo bgpgid=$(ps -o pgid= -p $! | tr -d ' '); \
             kill $! 2>/dev/null; wait 2>/dev/null; exit 0"
                .to_string(),
        ],
        SpawnBackend::Fork,
    )
    .pgroup(ProcessGroup::new(None, true))
    .session_containment()
    .capture_stdout()
    .capture_stderr()
    .timeout_ms(10_000)
    .kill_grace_ms(200)
    .cancel(CancelPolicy::Kill)
    .build()
    .unwrap()
    .run()
    .unwrap();

    let merged = String::from_utf8_lossy(&out.stdout);
    let shell_pgid = merged
        .lines()
        .find_map(|l| l.trim().strip_prefix("shell=").map(str::trim))
        .expect("shell pgid");
    let bg_pgid = merged
        .lines()
        .find_map(|l| l.trim().strip_prefix("bgpgid=").map(str::trim))
        .expect("bg pgid");
    assert_eq!(
        shell_pgid, bg_pgid,
        "pipe containment must still deny setpgid (job stays in shell's pgrp), got shell={shell_pgid} bg={bg_pgid}"
    );
    assert!(!out.timed_out, "pipe child must exit cleanly");
}

#[test]
fn test_pty_containment_still_blocks_session_escape_syscalls() {
    // The pty filter variant still denies the session-escape syscalls. Each
    // probe distinguishes seccomp from the natural error on this host:
    //   setsid: a one-fork-deep grandchild is not a pgrp leader, so uncontained
    //           it would succeed and escape (verified on-device and on host);
    //           under the filter it must report EPERM (errno 1).
    //   setns:  setns(99, 0) is EBADF (9) uncontained; EPERM (1) under the filter.
    //   unshare: unshare(CLONE_NEWUSER) succeeds uncontained as uid 1000;
    //           EPERM (1) under the filter.
    if !crate::fs::path_exists("/usr/bin/python3") {
        return;
    }

    let setsid_probe = concat!(
        "import os, time\n",
        "pid = os.fork()\n",
        "if pid == 0:\n",
        "    try:\n",
        "        os.setsid()\n",
        "        print('gc_setsid_ok', flush=True)\n",
        "    except OSError as e:\n",
        "        print('gc_setsid_errno=%d' % e.errno, flush=True)\n",
        "    time.sleep(1)\n",
        "else:\n",
        "    time.sleep(1)\n",
    );
    let out = SpawnOptions::builder(
        vec![
            "/usr/bin/python3".to_string(),
            "-c".to_string(),
            setsid_probe.to_string(),
        ],
        SpawnBackend::Fork,
    )
    .pgroup(ProcessGroup::new(None, true))
    .session_containment()
    .pty()
    .capture_stdout()
    .timeout_ms(10_000)
    .kill_grace_ms(200)
    .cancel(CancelPolicy::Kill)
    .build()
    .unwrap()
    .run()
    .unwrap();
    let merged = String::from_utf8_lossy(&out.stdout);
    assert!(
        merged.contains("gc_setsid_errno=1"),
        "pty containment must deny a grandchild's setsid, got {merged:?}"
    );
    assert!(
        !out.timed_out,
        "denied setsid must exit promptly, not hold the master"
    );

    let syscall_probes = r#"
import os
try:
    os.setns(99, 0); print("setns_ok")
except OSError as e:
    print("setns_errno=%d" % e.errno)
try:
    os.unshare(os.CLONE_NEWUSER); print("unshare_ok")
except OSError as e:
    print("unshare_errno=%d" % e.errno)
"#;
    let out = SpawnOptions::builder(
        vec![
            "/usr/bin/python3".to_string(),
            "-c".to_string(),
            syscall_probes.to_string(),
        ],
        SpawnBackend::Fork,
    )
    .pgroup(ProcessGroup::new(None, true))
    .session_containment()
    .pty()
    .capture_stdout()
    .timeout_ms(10_000)
    .kill_grace_ms(200)
    .cancel(CancelPolicy::Kill)
    .build()
    .unwrap()
    .run()
    .unwrap();
    let merged = String::from_utf8_lossy(&out.stdout);
    assert!(
        merged.contains("setns_errno=1"),
        "pty containment must deny setns (EPERM, not the natural EBADF), got {merged:?}"
    );
    assert!(
        merged.contains("unshare_errno=1"),
        "pty containment must deny unshare (EPERM, uncontained it succeeds), got {merged:?}"
    );
}

#[test]
fn test_pty_rejects_non_isolated_pgroup() {
    for backend in [
        SpawnBackend::Fork,
        SpawnBackend::Vfork,
        SpawnBackend::Clone3,
        SpawnBackend::Clone3Pidfd,
    ] {
        let opts = SpawnOptions::builder(vec!["/bin/true".to_string()], backend)
            .pty()
            .build()
            .unwrap();
        let err = match spawn_start(opts) {
            Ok(_) => panic!("{backend:?} pty must require an isolated pgroup"),
            Err(err) => err,
        };
        assert_eq!(err.raw_os_error(), Some(libc::EINVAL));
    }
}

#[test]
fn test_posix_spawn_rejects_pty() {
    let opts = SpawnOptions::builder(vec!["/bin/true".to_string()], SpawnBackend::PosixSpawn)
        .pgroup(ProcessGroup::new(None, true))
        .pty()
        .build()
        .unwrap();
    let err = match spawn_start(opts) {
        Ok(_) => panic!("expected posix_spawn pty to fail"),
        Err(err) => err,
    };
    assert_eq!(err.raw_os_error(), Some(libc::EINVAL));
}

#[test]
fn test_pty_rejects_stdin_buffer() {
    let opts = SpawnOptions::builder(vec!["/bin/true".to_string()], SpawnBackend::Fork)
        .pgroup(ProcessGroup::new(None, true))
        .pty()
        .stdin(b"hello".to_vec())
        .build()
        .unwrap();
    let err = match spawn_start(opts) {
        Ok(_) => panic!("pty + stdin buffer must be rejected"),
        Err(err) => err,
    };
    assert_eq!(err.raw_os_error(), Some(libc::EINVAL));
}

#[test]
fn test_pty_merged_io_and_controlling_terminal() {
    for backend in [SpawnBackend::Fork, SpawnBackend::Clone3Pidfd] {
        let script = "test -t 0 && test -t 1 && test -t 2 && echo stdio-is-tty; \
                      readlink /proc/self/fd/0; readlink /proc/self/fd/1; \
                      echo out-line; echo err-line >&2; \
                      echo sid=$(ps -o sid= -p $$); echo pgid=$(ps -o pgid= -p $$)";
        let out = SpawnOptions::builder(
            vec!["/bin/sh".to_string(), "-c".to_string(), script.to_string()],
            backend,
        )
        .pgroup(ProcessGroup::new(None, true))
        .pty()
        .capture_stdout()
        .build()
        .unwrap()
        .run()
        .unwrap();

        assert!(
            matches!(out.status, Some(ExitStatus::Exited(0))),
            "{backend:?}: pty child must exit 0, got {out:?}"
        );
        let merged = String::from_utf8_lossy(&out.stdout);
        assert!(
            merged.contains("stdio-is-tty"),
            "{backend:?}: all of stdin/stdout/stderr must be a tty, got {merged:?}"
        );
        let pts_lines: Vec<_> = merged
            .lines()
            .filter(|l| l.starts_with("/dev/pts/"))
            .collect();
        assert!(
            pts_lines.len() >= 2,
            "{backend:?}: fd 0/1 must be a pty slave, got {merged:?}"
        );
        let out_pos = merged.find("out-line").expect("merged stdout");
        let err_pos = merged.find("err-line").expect("merged stderr");
        assert!(
            out_pos < err_pos,
            "{backend:?}: stdout/stderr must be merged in order, got {merged:?}"
        );
        // The child is a session leader with its own group: sid == pid == pgid.
        // (`ps -o sid=` right-pads to width 5, so compare trimmed values.)
        let sid = merged
            .lines()
            .find_map(|l| l.trim_start().strip_prefix("sid="))
            .expect("sid line")
            .trim();
        let pgid = merged
            .lines()
            .find_map(|l| l.trim_start().strip_prefix("pgid="))
            .expect("pgid line")
            .trim();
        assert_eq!(
            sid, pgid,
            "{backend:?}: child must be a session+group leader, got {merged:?}"
        );
        assert_eq!(
            sid,
            out.pid.to_string(),
            "{backend:?}: child must be its own session leader (sid==pid), got {merged:?}"
        );
    }
}

#[test]
fn test_pty_resize_applies_winsize() {
    for backend in [SpawnBackend::Fork, SpawnBackend::Clone3Pidfd] {
        let managed = spawn_managed(
            SpawnOptions::builder(
                vec![
                    "/bin/sh".to_string(),
                    "-c".to_string(),
                    "sleep 1; stty size".to_string(),
                ],
                backend,
            )
            .pgroup(ProcessGroup::new(None, true))
            .pty()
            .capture_stdout()
            .timeout_ms(10_000)
            .kill_grace_ms(200)
            .cancel(CancelPolicy::Kill)
            .build()
            .unwrap(),
        )
        .unwrap();

        // Resize before the child's `sleep 1` elapses so its `stty size`
        // observes the new window.
        managed.resize_pty(24, 100).unwrap();

        let output = drive_managed(managed, false).unwrap();
        assert!(!output.timed_out, "{backend:?}: pty resize child must exit");
        let merged = String::from_utf8_lossy(&output.stdout);
        assert!(
            merged.trim() == "24 100",
            "{backend:?}: TIOCSWINSZ must propagate to the child, got {merged:?}"
        );
    }
}

#[test]
fn test_pty_with_caller_pair_window_and_env() {
    for backend in [SpawnBackend::Fork, SpawnBackend::Clone3Pidfd] {
        let (master, slave) = make_pty().expect("make_pty");
        let (rows, cols) = pty_window(&master, 30, 120).expect("pty_window");
        assert_eq!((rows, cols), (30, 120), "{backend:?}: winsize read-back");

        // The caller derives LINES/COLUMNS from the read-back and passes them
        // through the builder env, exactly like the daemon's pty mode does.
        let managed = spawn_managed(
            SpawnOptions::builder(
                vec![
                    "/bin/sh".to_string(),
                    "-c".to_string(),
                    "echo \"$LINES:$COLUMNS\"; stty size".to_string(),
                ],
                backend,
            )
            .env(vec![format!("LINES={rows}"), format!("COLUMNS={cols}")])
            .pgroup(ProcessGroup::new(None, true))
            .pty_with(master, slave)
            .capture_stdout()
            .timeout_ms(10_000)
            .kill_grace_ms(200)
            .cancel(CancelPolicy::Kill)
            .build()
            .unwrap(),
        )
        .unwrap();

        let output = drive_managed(managed, false).unwrap();
        assert!(!output.timed_out, "{backend:?}: pty_with child must exit");
        let merged = String::from_utf8_lossy(&output.stdout);
        let env_line = merged.lines().next().unwrap_or("");
        let stty_line = merged.lines().nth(1).unwrap_or("");
        assert_eq!(
            env_line, "30:120",
            "{backend:?}: env LINES/COLUMNS must match the applied window, got {merged:?}"
        );
        assert_eq!(
            stty_line, "30 120",
            "{backend:?}: child must observe the applied winsize, got {merged:?}"
        );
    }
}

#[test]
fn test_pty_without_env_inherits_winsize() {
    for backend in [SpawnBackend::Fork, SpawnBackend::Clone3Pidfd] {
        let (master, slave) = make_pty().expect("make_pty");
        let (rows, cols) = pty_window(&master, 40, 80).expect("pty_window");
        assert_eq!((rows, cols), (40, 80), "{backend:?}: winsize read-back");

        let managed = spawn_managed(
            SpawnOptions::builder(
                vec![
                    "/bin/sh".to_string(),
                    "-c".to_string(),
                    // No env override: the shell derives LINES/COLUMNS from the
                    // terminal when it initializes.
                    "stty size; sh -c 'echo ${LINES:-unset}:${COLUMNS:-unset}'".to_string(),
                ],
                backend,
            )
            .pgroup(ProcessGroup::new(None, true))
            .pty_with(master, slave)
            .capture_stdout()
            .timeout_ms(10_000)
            .kill_grace_ms(200)
            .cancel(CancelPolicy::Kill)
            .build()
            .unwrap(),
        )
        .unwrap();

        let output = drive_managed(managed, false).unwrap();
        assert!(!output.timed_out, "{backend:?}: pty_with child must exit");
        let merged = String::from_utf8_lossy(&output.stdout);
        let stty_line = merged.lines().next().unwrap_or("");
        assert_eq!(
            stty_line, "40 80",
            "{backend:?}: child must observe the applied winsize, got {merged:?}"
        );
    }
}

#[test]
fn test_pty_write_input_round_trips_to_the_child() {
    for backend in [SpawnBackend::Fork, SpawnBackend::Clone3Pidfd] {
        let managed = spawn_managed(
            SpawnOptions::builder(
                vec![
                    "/bin/sh".to_string(),
                    "-c".to_string(),
                    "read line; echo got:$line".to_string(),
                ],
                backend,
            )
            .pgroup(ProcessGroup::new(None, true))
            .pty()
            .capture_stdout()
            .timeout_ms(10_000)
            .kill_grace_ms(200)
            .cancel(CancelPolicy::Kill)
            .build()
            .unwrap(),
        )
        .unwrap();

        // Bytes written to the master reach the child's stdin; the tty line
        // discipline (ICRNL) turns the CR into the newline `read` waits on.
        let n = managed.write_input(b"hello\r").unwrap();
        assert_eq!(n, 6, "{backend:?}: write_input must accept every byte");

        let output = drive_managed(managed, false).unwrap();
        assert!(!output.timed_out, "{backend:?}: echo child must exit");
        let merged = String::from_utf8_lossy(&output.stdout);
        assert!(
            merged.contains("got:hello"),
            "{backend:?}: child must read the written line, got {merged:?}"
        );
    }
}

#[test]
fn test_pty_write_input_ctrl_c_signals_the_foreground_group() {
    for backend in [SpawnBackend::Fork, SpawnBackend::Clone3Pidfd] {
        let managed = spawn_managed(
            SpawnOptions::builder(
                vec![
                    "/bin/sh".to_string(),
                    "-c".to_string(),
                    "echo start; sleep 30".to_string(),
                ],
                backend,
            )
            .pgroup(ProcessGroup::new(None, true))
            .pty()
            .capture_stdout()
            .timeout_ms(10_000)
            .kill_grace_ms(200)
            .cancel(CancelPolicy::Kill)
            .build()
            .unwrap(),
        )
        .unwrap();

        // With ISIG on (the default), the line discipline turns 0x03 into
        // SIGINT to the tty's foreground process group — the child itself.
        managed.write_input(b"\x03").unwrap();

        let output = drive_managed(managed, false).unwrap();
        assert!(
            !output.timed_out,
            "{backend:?}: Ctrl-C must preempt the sleep before the timeout"
        );
        let status = output.status;
        assert!(
            matches!(
                status,
                Some(ExitStatus::Signaled(libc::SIGINT)) | Some(ExitStatus::Exited(130))
            ),
            "{backend:?}: Ctrl-C must terminate the child by SIGINT, got {status:?}"
        );
    }
}

#[test]
fn test_pty_write_input_rejects_non_pty_spawn() {
    let managed = spawn_managed(
        SpawnOptions::builder(
            vec!["/bin/sleep".to_string(), "0.2".to_string()],
            SpawnBackend::Fork,
        )
        .capture_stdout()
        .build()
        .unwrap(),
    )
    .unwrap();
    match managed.write_input(b"x") {
        Err(e) => assert_eq!(e.raw_os_error(), Some(libc::EINVAL)),
        Ok(n) => panic!("write_input on a pipe-capture spawn must be refused, wrote {n}"),
    }
}

#[test]
fn test_pty_write_input_nonblock_round_trips_to_the_child() {
    // D4-6: the non-blocking pty write must accept a line without blocking and
    // the child must read it (mirrors the blocking write_input round-trip).
    for backend in [SpawnBackend::Fork, SpawnBackend::Clone3Pidfd] {
        let managed = spawn_managed(
            SpawnOptions::builder(
                vec![
                    "/bin/sh".to_string(),
                    "-c".to_string(),
                    "read line; echo got:$line".to_string(),
                ],
                backend,
            )
            .pgroup(ProcessGroup::new(None, true))
            .pty()
            .capture_stdout()
            .timeout_ms(10_000)
            .kill_grace_ms(200)
            .cancel(CancelPolicy::Kill)
            .build()
            .unwrap(),
        )
        .unwrap();

        // A fresh pty input buffer accepts the whole line without blocking.
        let n = managed.write_input_nonblock(b"hello\r").unwrap();
        assert_eq!(
            n,
            Some(6),
            "{backend:?}: write_input_nonblock must accept every byte on a fresh pty"
        );

        let output = drive_managed(managed, false).unwrap();
        assert!(!output.timed_out, "{backend:?}: echo child must exit");
        let merged = String::from_utf8_lossy(&output.stdout);
        assert!(
            merged.contains("got:hello"),
            "{backend:?}: child must read the written line, got {merged:?}"
        );
    }
}

#[test]
fn test_drain_maps_pty_master_eio_to_eof() {
    // A pty master returns EIO (not EOF) once all slave holders close. The
    // drain must map that to a clean EOF so pty jobs complete instead of
    // surfacing a spurious I/O error.
    let master = unsafe {
        libc::open(
            c"/dev/ptmx".as_ptr(),
            libc::O_RDWR | libc::O_NOCTTY | libc::O_CLOEXEC,
        )
    };
    assert!(master >= 0);
    assert_eq!(unsafe { libc::grantpt(master) }, 0);
    assert_eq!(unsafe { libc::unlockpt(master) }, 0);
    let mut name = [0 as libc::c_char; 4096];
    assert_eq!(
        unsafe { libc::ptsname_r(master, name.as_mut_ptr(), name.len()) },
        0
    );
    let slave = unsafe {
        libc::open(
            name.as_ptr(),
            libc::O_RDWR | libc::O_NOCTTY | libc::O_CLOEXEC,
        )
    };
    assert!(slave >= 0);

    let master_fd = Fd::new(master, "test pty master").unwrap();
    let slave_fd = Fd::new(slave, "test pty slave").unwrap();

    // First read: nothing written yet, non-blocking EAGAIN -> Open (no EOF).
    let mut drain: DrainState<fn(&[u8]) -> bool> =
        DrainState::new(None, None, Some(master_fd), None, 1024, None, None, true).unwrap();
    assert!(!drain.read_fd(true).unwrap());

    // Child "exits": close the slave. The master then returns EIO.
    drop(slave_fd);
    let mut reads = 0;
    while !drain.read_fd(true).unwrap() {
        reads += 1;
        if reads > 100 {
            panic!("pty master EIO was not mapped to EOF");
        }
    }
    assert!(drain.is_done());

    // Without the pty flag the same EIO surfaces as an error.
    let master2 = unsafe {
        libc::open(
            c"/dev/ptmx".as_ptr(),
            libc::O_RDWR | libc::O_NOCTTY | libc::O_CLOEXEC,
        )
    };
    assert!(master2 >= 0);
    assert_eq!(unsafe { libc::grantpt(master2) }, 0);
    assert_eq!(unsafe { libc::unlockpt(master2) }, 0);
    let mut name2 = [0 as libc::c_char; 4096];
    assert_eq!(
        unsafe { libc::ptsname_r(master2, name2.as_mut_ptr(), name2.len()) },
        0
    );
    let slave2 = unsafe {
        libc::open(
            name2.as_ptr(),
            libc::O_RDWR | libc::O_NOCTTY | libc::O_CLOEXEC,
        )
    };
    assert!(slave2 >= 0);
    let master2_fd = Fd::new(master2, "test pty master 2").unwrap();
    let slave2_fd = Fd::new(slave2, "test pty slave 2").unwrap();
    let mut drain2: DrainState<fn(&[u8]) -> bool> =
        DrainState::new(None, None, Some(master2_fd), None, 1024, None, None, false).unwrap();
    drop(slave2_fd);
    let err = loop {
        match drain2.read_fd(true) {
            Ok(true) => panic!("non-pty drain must not treat EIO as EOF"),
            Ok(false) => continue,
            Err(e) => break e,
        }
    };
    assert_eq!(err.raw_os_error(), Some(libc::EIO));
}

#[test]
fn test_vfork_reports_child_chdir_error() {
    let err = SpawnOptions::builder(vec!["/bin/true".to_string()], SpawnBackend::Vfork)
        .cwd("/definitely/missing/coreshift-core-test".to_string())
        .build()
        .unwrap()
        .run()
        .unwrap_err();
    assert_eq!(err.raw_os_error(), Some(libc::ENOENT));
    assert!(err.to_string().contains("spawn child chdir"));
}

#[test]
fn test_clone3_reports_child_chdir_error() {
    let err = SpawnOptions::builder(vec!["/bin/true".to_string()], SpawnBackend::Clone3)
        .cwd("/definitely/missing/coreshift-core-test".to_string())
        .build()
        .unwrap()
        .run()
        .unwrap_err();
    assert_eq!(err.raw_os_error(), Some(libc::ENOENT));
    assert!(err.to_string().contains("spawn child chdir"));
}

#[test]
fn test_fork_close_from3_closes_many_inherited_fds() {
    // More fds than fit in one 4096-byte getdents64 batch (~170 entries): the
    // child scan must keep reading batches through EOF, not stop at the first
    // short read, or every fd in a later batch would leak into the child.
    let mut files = Vec::new();
    let mut fds = Vec::new();
    for _ in 0..200 {
        let f = File::open("/dev/null").unwrap();
        let fd = f.as_raw_fd();
        let flags = unsafe { libc::fcntl(fd, libc::F_GETFD) };
        assert!(flags >= 0);
        let ret = unsafe { libc::fcntl(fd, libc::F_SETFD, flags & !libc::FD_CLOEXEC) };
        assert_eq!(ret, 0);
        fds.push(fd);
        files.push(f);
    }
    // A probe fd guaranteed to land in a later batch (highest number).
    let probe = *fds.last().unwrap();

    let script = format!("if [ -e /proc/$$/fd/{probe} ]; then echo open; else echo closed; fi");
    let output = SpawnOptions::builder(
        vec!["/bin/sh".to_string(), "-c".to_string(), script],
        SpawnBackend::Fork,
    )
    .fd_policy(SpawnFdPolicy::CloseFrom3)
    .capture_stdout()
    .build()
    .unwrap()
    .run()
    .unwrap();

    assert_eq!(String::from_utf8_lossy(&output.stdout).trim(), "closed");
    drop(files);
}

#[test]
fn test_fork_allowlist_preserves_allowed_fd() {
    let file = File::open("/dev/null").unwrap();
    let fd = file.as_raw_fd();
    let flags = unsafe { libc::fcntl(fd, libc::F_GETFD) };
    assert!(flags >= 0);
    let ret = unsafe { libc::fcntl(fd, libc::F_SETFD, flags & !libc::FD_CLOEXEC) };
    assert_eq!(ret, 0);

    let script = format!("if [ -e /proc/$$/fd/{fd} ]; then echo open; else echo closed; fi");
    let output = SpawnOptions::builder(
        vec!["/bin/sh".to_string(), "-c".to_string(), script],
        SpawnBackend::Fork,
    )
    .fd_policy(SpawnFdPolicy::Allowlist(vec![fd]))
    .capture_stdout()
    .build()
    .unwrap()
    .run()
    .unwrap();

    assert_eq!(String::from_utf8_lossy(&output.stdout).trim(), "open");
}

#[test]
fn test_process_echild() {
    // Use an invalid PID that likely doesn't exist and isn't our child.
    let p = Process::new(999999);
    let res = p.wait_step();
    // Should be an error (ECHILD), not Ok(Some(Exited(0))).
    assert!(res.is_err());
}

#[test]
fn test_spawn_start_wait_false_validation() {
    // Valid: wait=false, no I/O capture
    assert!(spawn_start(base_wait_false_builder().build().unwrap()).is_ok());

    // Invalid: wait=false, capture_stdout=true
    assert!(spawn_start(base_wait_false_builder().capture_stdout().build().unwrap()).is_err());

    // Invalid: wait=false, stdin=Some(...)
    assert!(
        spawn_start(
            base_wait_false_builder()
                .stdin(vec![1, 2, 3])
                .build()
                .unwrap()
        )
        .is_err()
    );
}

/// The shared `wait=false` builder for the validation test above. Rebuilt per
/// assertion: the builder is move-only once it owns a pty pair (and not
/// `Clone` at all — see [`SpawnOptions`]).
fn base_wait_false_builder() -> SpawnOptionsBuilder {
    SpawnOptions::builder(
        vec!["/bin/sh".to_string(), "-c".to_string(), "true".to_string()],
        SpawnBackend::PosixSpawn,
    )
    .wait(false)
    .max_output(1024)
    .kill_grace_ms(1000)
}

#[test]
fn test_spawn_wait_false_child_is_reaped_by_reaper() {
    // A `wait=false` child is never waited on by the caller. Once it exits it
    // must still be reaped (by the orphan reaper) or it becomes a zombie in a
    // long-lived daemon (finding 15). Give the reaper its 250 ms tick to run.
    let output = SpawnOptions::builder(vec!["/bin/true".to_string()], SpawnBackend::PosixSpawn)
        .wait(false)
        .build()
        .unwrap()
        .run()
        .unwrap();

    let pid = output.pid;
    assert!(pid > 0);

    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
    loop {
        let mut status: libc::c_int = 0;
        let r = unsafe { libc::waitpid(pid, &mut status, libc::WNOHANG) };
        // The child has been reaped (either by the reaper or our own waitpid).
        if r == pid {
            break;
        }
        // ECHILD: already reaped by the background reaper — success.
        if r < 0 && std::io::Error::last_os_error().raw_os_error() == Some(libc::ECHILD) {
            return;
        }
        if std::time::Instant::now() > deadline {
            panic!("wait=false child {pid} was never reaped (zombie leak)");
        }
        std::thread::sleep(std::time::Duration::from_millis(50));
    }
}

#[test]
fn test_pdeath_signal_kills_child_when_parent_exits() {
    // D3: `pdeath_signal(SIGKILL)` arms PR_SET_PDEATHSIG in the child. When
    // the parent (the process that spawned it) exits, the child must be killed.
    // The test forks a short-lived "daemon" that spawns a long-lived child with
    // the flag; when the daemon exits, the child must die with SIGKILL.
    use std::io::Read;

    let mut fds = [0i32; 2];
    assert_eq!(unsafe { libc::pipe(fds.as_mut_ptr()) }, 0);
    // CLOEXEC so the daemon's spawned grandchild does not inherit the write
    // end (default spawn fd policy keeps fds beyond 0/1/2 unless CloseFrom3);
    // otherwise the parent's read below never sees EOF.
    for f in fds {
        let fl = unsafe { libc::fcntl(f, libc::F_GETFD) };
        assert!(fl >= 0);
        assert_eq!(
            unsafe { libc::fcntl(f, libc::F_SETFD, fl | libc::FD_CLOEXEC) },
            0
        );
    }
    let (r, w) = (fds[0], fds[1]);
    let daemon = unsafe { libc::fork() };
    assert!(daemon >= 0);
    if daemon == 0 {
        // Child (daemon): spawn a grandchild with pdeath_signal, report its
        // pid, then exit. The grandchild's parent thread (this process) dies,
        // so PR_SET_PDEATHSIG must deliver SIGKILL to it.
        let opts = SpawnOptions::builder(
            vec!["/bin/sleep".to_string(), "1000".to_string()],
            SpawnBackend::Fork,
        )
        .wait(false)
        .pdeath_signal(libc::SIGKILL)
        .build()
        .unwrap();
        let running = spawn_start(opts).unwrap();
        let gp = running.process.pid();
        unsafe {
            let pid_str = gp.to_string();
            let bytes = pid_str.as_bytes();
            let mut written = 0;
            while written < bytes.len() {
                let n = libc::write(w, bytes[written..].as_ptr().cast(), bytes.len() - written);
                if n < 0 {
                    if std::io::Error::last_os_error().raw_os_error() == Some(libc::EINTR) {
                        continue;
                    }
                    break;
                }
                written += n as usize;
            }
            libc::_exit(0);
        }
    }

    // Parent (test process): read the grandchild pid, reap the daemon, then
    // poll for the grandchild's death (pdeath delivered).
    // Parent: close the write end so the read below sees EOF once the daemon
    // (and its exec'd grandchild, whose fds are CloseFrom3'd) have exited.
    unsafe { libc::close(w) };
    let mut pid_buf = String::new();
    {
        use std::os::unix::io::FromRawFd;
        let rfile = unsafe { std::fs::File::from_raw_fd(r) };
        let mut br = std::io::BufReader::new(rfile);
        br.read_to_string(&mut pid_buf).unwrap();
    }
    let gp: i32 = pid_buf.trim().parse().unwrap();
    assert!(gp > 0);

    let mut status: libc::c_int = 0;
    assert_eq!(unsafe { libc::waitpid(daemon, &mut status, 0) }, daemon);

    // The grandchild must disappear within a bounded window (pdeath SIGKILL).
    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
    loop {
        let alive = std::path::Path::new(&format!("/proc/{gp}")).exists();
        if !alive {
            break;
        }
        if std::time::Instant::now() > deadline {
            // Clean up the leak before panicking.
            unsafe { libc::kill(gp, libc::SIGKILL) };
            panic!("grandchild {gp} survived the daemon's exit — pdeath_signal not delivered");
        }
        std::thread::sleep(std::time::Duration::from_millis(50));
    }
}

#[test]
fn test_pdeath_signal_off_means_child_survives_parent_exit() {
    // Negative control for D3: without `pdeath_signal`, a wait=false child
    // must survive its parent's exit (standard background-child semantics).
    use std::io::Read;

    let mut fds = [0i32; 2];
    assert_eq!(unsafe { libc::pipe(fds.as_mut_ptr()) }, 0);
    // CLOEXEC so the daemon's spawned grandchild does not inherit the write
    // end (default spawn fd policy keeps fds beyond 0/1/2 unless CloseFrom3);
    // otherwise the parent's read below never sees EOF.
    for f in fds {
        let fl = unsafe { libc::fcntl(f, libc::F_GETFD) };
        assert!(fl >= 0);
        assert_eq!(
            unsafe { libc::fcntl(f, libc::F_SETFD, fl | libc::FD_CLOEXEC) },
            0
        );
    }
    let (r, w) = (fds[0], fds[1]);
    let daemon = unsafe { libc::fork() };
    assert!(daemon >= 0);
    if daemon == 0 {
        let opts = SpawnOptions::builder(
            vec!["/bin/sleep".to_string(), "1000".to_string()],
            SpawnBackend::Fork,
        )
        .wait(false)
        .build()
        .unwrap();
        let running = spawn_start(opts).unwrap();
        let gp = running.process.pid();
        unsafe {
            let pid_str = gp.to_string();
            let bytes = pid_str.as_bytes();
            let mut written = 0;
            while written < bytes.len() {
                let n = libc::write(w, bytes[written..].as_ptr().cast(), bytes.len() - written);
                if n < 0 {
                    if std::io::Error::last_os_error().raw_os_error() == Some(libc::EINTR) {
                        continue;
                    }
                    break;
                }
                written += n as usize;
            }
            libc::_exit(0);
        }
    }

    // Parent: close the write end so the read below sees EOF once the daemon
    // (and its exec'd grandchild, whose fds are CloseFrom3'd) have exited.
    unsafe { libc::close(w) };
    let mut pid_buf = String::new();
    {
        use std::os::unix::io::FromRawFd;
        let rfile = unsafe { std::fs::File::from_raw_fd(r) };
        let mut br = std::io::BufReader::new(rfile);
        br.read_to_string(&mut pid_buf).unwrap();
    }
    let gp: i32 = pid_buf.trim().parse().unwrap();
    assert!(gp > 0);

    let mut status: libc::c_int = 0;
    assert_eq!(unsafe { libc::waitpid(daemon, &mut status, 0) }, daemon);

    // The child must still be alive after a short settle window.
    std::thread::sleep(std::time::Duration::from_millis(300));
    assert!(
        std::path::Path::new(&format!("/proc/{gp}")).exists(),
        "child {gp} died without pdeath_signal"
    );
    // Clean up: kill + reap the leaked background child.
    unsafe {
        libc::kill(gp, libc::SIGKILL);
        libc::waitpid(gp, &mut status, 0);
    }
}

#[test]
fn test_pdeath_signal_rejected_on_posix_spawn() {
    // posix_spawn has no child-setup step, so PR_SET_PDEATHSIG cannot be
    // armed there; the backend validator must reject the combination.
    let err = SpawnOptions::builder(vec!["/bin/true".to_string()], SpawnBackend::PosixSpawn)
        .pdeath_signal(libc::SIGKILL)
        .build()
        .unwrap()
        .run()
        .unwrap_err();
    assert_eq!(err.raw_os_error(), Some(libc::EINVAL));
}

#[test]
fn test_reactor_wait_zero_events() {
    let mut reactor = Reactor::new().unwrap();
    let mut events = Vec::new();
    let res = reactor.wait(&mut events, 0, 0);
    assert!(res.is_ok());
    assert_eq!(res.unwrap(), 0);
}

#[test]
fn test_reactor_add_priority_registers_fd() {
    let mut reactor = Reactor::new().unwrap();
    let fd = Fd::eventfd(0).unwrap();
    let token = reactor.add_priority(&fd).unwrap();
    let mut events = Vec::new();

    assert_eq!(reactor.wait(&mut events, 4, 0).unwrap(), 0);
    assert!(events.is_empty());
    assert!(format!("{token:?}").starts_with("Token("));
}

#[test]
fn test_reactor_mod_toggles_read_interest() {
    let mut reactor = Reactor::new().unwrap();
    let fd = Fd::eventfd(0).unwrap();
    let token = reactor.add(&fd, true, false).unwrap();
    let mut events = Vec::new();

    // eventfd starts with counter 0: not readable, no event.
    assert_eq!(reactor.wait(&mut events, 4, 0).unwrap(), 0);

    // Counter becomes readable → edge fires.
    fd.write_u64(1).unwrap();
    events.clear();
    assert_eq!(reactor.wait(&mut events, 4, 100).unwrap(), 1);
    assert_eq!(events[0].token, token);
    assert!(events[0].readable);

    // Drain it.
    assert_eq!(fd.read_u64().unwrap(), Some(1));

    // MOD: clear the readable interest (direction-preserving pause). Data
    // that arrives while the interest is cleared must not fire.
    reactor.mod_(&fd, token, false, false).unwrap();
    fd.write_u64(1).unwrap();
    events.clear();
    assert_eq!(
        reactor.wait(&mut events, 4, 50).unwrap(),
        0,
        "cleared readable interest must not deliver events"
    );

    // MOD: re-arm readable interest. EPOLL_CTL_MOD re-evaluates readiness, so
    // the pending byte fires again on the SAME token.
    reactor.mod_(&fd, token, true, false).unwrap();
    events.clear();
    assert_eq!(reactor.wait(&mut events, 4, 100).unwrap(), 1);
    assert_eq!(events[0].token, token);
    assert!(events[0].readable);
}

#[test]
fn test_fd_slice_io_helpers() {
    let mut fds = [0; 2];
    unsafe { libc::pipe(fds.as_mut_ptr()) };
    let r = Fd::new(fds[0], "pipe").unwrap();
    let w = Fd::new(fds[1], "pipe").unwrap();

    let input = b"hello";
    let written = w.write_slice(input).unwrap().unwrap();
    assert_eq!(written, input.len());

    let mut buf = [0u8; 8];
    let read = r.read_slice(&mut buf).unwrap().unwrap();
    assert_eq!(read, input.len());
    assert_eq!(&buf[..read], input);
}

#[test]
fn test_unix_socket_bind_connect_roundtrip() {
    let path = temp_socket_path("unix_roundtrip");
    let listener = bind_test_unix_listener(&path, false);
    let client = connect_test_unix_stream(UnixSocketAddr::Path(&path));

    let server = loop {
        if let Some(server) = listener.accept().unwrap() {
            break server;
        }
        std::thread::yield_now();
    };

    let written = client.fd().write_slice(b"ping").unwrap().unwrap();
    assert_eq!(written, 4);

    let mut buf = [0u8; 8];
    let read = loop {
        if let Some(read) = server.fd().read_slice(&mut buf).unwrap() {
            break read;
        }
        std::thread::yield_now();
    };
    assert_eq!(&buf[..read], b"ping");

    chmod(UnixSocketAddr::Path(&path), 0o600).unwrap();
    let mode = std::fs::metadata(&path).unwrap().permissions().mode();
    assert_eq!(mode & 0o777, 0o600);

    let _ = remove_file(&path);
}

#[test]
fn test_unix_socket_abstract_bind_connect_roundtrip() {
    let name = format!(
        "coreshift_abstract_{}_{}",
        std::process::id(),
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_nanos()
    );
    let listener = bind(
        UnixSocketAddr::Abstract(name.as_bytes()),
        UnixSocketBindOptions::default(),
    )
    .unwrap();
    let client = connect_test_unix_stream(UnixSocketAddr::Abstract(name.as_bytes()));

    let server = loop {
        if let Some(server) = listener.accept().unwrap() {
            break server;
        }
        std::thread::yield_now();
    };

    let written = client.fd().write_slice(b"pong").unwrap().unwrap();
    assert_eq!(written, 4);

    let mut buf = [0u8; 8];
    let read = loop {
        if let Some(read) = server.fd().read_slice(&mut buf).unwrap() {
            break read;
        }
        std::thread::yield_now();
    };
    assert_eq!(&buf[..read], b"pong");
}

#[test]
fn test_unix_socket_connect_result_finish_is_safe() {
    let path = temp_socket_path("unix_connect_result");
    let _listener = bind_test_unix_listener(&path, false);

    let client = match connect(UnixSocketAddr::Path(&path)).unwrap() {
        UnixConnectResult::Connected(stream) => {
            assert_eq!(stream.check_connect_error().unwrap(), None);
            stream
        }
        UnixConnectResult::InProgress(stream) => stream.finish_connect().unwrap(),
    };

    drop(client);
    let _ = remove_file(&path);
}

#[test]
fn test_unix_socket_unlinks_stale_path_when_requested() {
    let path = temp_socket_path("unix_unlink_stale");
    {
        let _listener = bind_test_unix_listener(&path, false);
    }

    let _listener = bind_test_unix_listener(&path, true);
    assert!(std::fs::metadata(&path).unwrap().file_type().is_socket());

    let _ = remove_file(&path);
}

#[test]
fn test_unix_socket_does_not_unlink_stale_path_by_default() {
    let path = temp_socket_path("unix_no_unlink");
    File::create(&path).unwrap();

    let result = bind(
        UnixSocketAddr::Path(&path),
        UnixSocketBindOptions {
            stale_socket_policy: StaleSocketPolicy::Preserve,
            mode: None,
        },
    );
    let err = match result {
        Ok(_) => panic!("bind unexpectedly succeeded"),
        Err(err) => err,
    };
    assert_eq!(err.raw_os_error(), Some(libc::EADDRINUSE));

    let _ = remove_file(&path);
}

#[test]
fn test_unix_socket_unlink_socket_only_preserves_regular_file() {
    let path = temp_socket_path("unix_regular_preserved");
    File::create(&path).unwrap();

    let result = bind(
        UnixSocketAddr::Path(&path),
        UnixSocketBindOptions {
            stale_socket_policy: StaleSocketPolicy::UnlinkSocketOnly,
            mode: None,
        },
    );
    let err = match result {
        Ok(_) => panic!("bind unexpectedly succeeded"),
        Err(err) => err,
    };
    assert_eq!(err.raw_os_error(), Some(libc::EEXIST));
    assert!(std::fs::metadata(&path).unwrap().is_file());

    let _ = remove_file(&path);
}

#[test]
fn test_unix_socket_nul_path_rejected() {
    let path = PathBuf::from(std::ffi::OsString::from_vec(b"/tmp/core\0socket".to_vec()));
    let result = bind(
        UnixSocketAddr::Path(&path),
        UnixSocketBindOptions {
            stale_socket_policy: StaleSocketPolicy::Preserve,
            mode: None,
        },
    );
    let err = match result {
        Ok(_) => panic!("bind unexpectedly succeeded"),
        Err(err) => err,
    };
    assert_eq!(err.raw_os_error(), Some(libc::EINVAL));

    let path = PathBuf::from(std::ffi::OsString::from_vec(b"/tmp/core\0socket".to_vec()));
    let result = connect(UnixSocketAddr::Path(&path));
    let err = match result {
        Ok(_) => panic!("connect unexpectedly succeeded"),
        Err(err) => err,
    };
    assert_eq!(err.raw_os_error(), Some(libc::EINVAL));
}

#[test]
fn test_unix_socket_overlong_path_and_abstract_rejected() {
    let path = PathBuf::from("/tmp".to_string() + &"/x".repeat(120));
    let result = bind(
        UnixSocketAddr::Path(&path),
        UnixSocketBindOptions::default(),
    );
    let err = match result {
        Ok(_) => panic!("bind unexpectedly succeeded"),
        Err(err) => err,
    };
    assert_eq!(err.raw_os_error(), Some(libc::ENAMETOOLONG));

    let name = vec![b'x'; 108];
    let result = bind(
        UnixSocketAddr::Abstract(&name),
        UnixSocketBindOptions::default(),
    );
    let err = match result {
        Ok(_) => panic!("bind unexpectedly succeeded"),
        Err(err) => err,
    };
    assert_eq!(err.raw_os_error(), Some(libc::ENAMETOOLONG));
}

#[test]
fn test_unix_socket_accept_returns_none_on_eagain() {
    let path = temp_socket_path("unix_accept_eagain");
    let listener = bind_test_unix_listener(&path, false);

    assert!(listener.accept().unwrap().is_none());

    let _ = remove_file(&path);
}

#[test]
fn test_unix_socket_chmod_path_only() {
    let regular = temp_socket_path("unix_chmod_regular");
    File::create(&regular).unwrap();
    let err = chmod(UnixSocketAddr::Path(&regular), 0o600).unwrap_err();
    assert_eq!(err.raw_os_error(), Some(libc::EINVAL));
    let _ = remove_file(&regular);

    let name = b"coreshift_abstract_chmod_rejected";
    let err = chmod(UnixSocketAddr::Abstract(name), 0o600).unwrap_err();
    assert_eq!(err.raw_os_error(), Some(libc::EINVAL));

    let result = bind(
        UnixSocketAddr::Abstract(name),
        UnixSocketBindOptions {
            stale_socket_policy: StaleSocketPolicy::Preserve,
            mode: Some(0o600),
        },
    );
    let err = match result {
        Ok(_) => panic!("bind unexpectedly succeeded"),
        Err(err) => err,
    };
    assert_eq!(err.raw_os_error(), Some(libc::EINVAL));
}

#[test]
fn test_unix_socket_peer_credentials_when_supported() {
    let path = temp_socket_path("unix_peer_cred");
    let listener = bind_test_unix_listener(&path, false);
    let client = connect_test_unix_stream(UnixSocketAddr::Path(&path));

    let server = loop {
        if let Some(server) = listener.accept().unwrap() {
            break server;
        }
        std::thread::yield_now();
    };

    if let Some(cred) = server.peer_cred().unwrap() {
        assert_eq!(cred.pid, Some(std::process::id() as i32));
        assert_eq!(cred.uid, unsafe { libc::geteuid() });
        assert_eq!(cred.gid, unsafe { libc::getegid() });
    }

    drop(client);
    let _ = remove_file(&path);
}

fn early_exit_on_stop(chunk: &[u8]) -> bool {
    chunk.windows(4).any(|window| window == b"stop")
}

#[test]
fn test_drain_early_exit_is_explicit_state_not_eof() {
    let mut fds = [0; 2];
    unsafe { libc::pipe2(fds.as_mut_ptr(), libc::O_CLOEXEC | libc::O_NONBLOCK) };
    let r = Fd::new(fds[0], "pipe").unwrap();
    let w = Fd::new(fds[1], "pipe").unwrap();
    w.write_slice(b"stop").unwrap();

    let mut drain = DrainState::new(
        None,
        None,
        Some(r),
        None,
        128,
        Some(early_exit_on_stop),
        None,
        false,
    )
    .unwrap();
    assert!(drain.read_fd(true).unwrap());
    assert!(drain.stdout_early_exited());

    assert!(!drain.output_limit_exceeded());
}

#[test]
fn test_drain_pause_preserves_registration_and_resume_rearms() {
    // D4-4: pause must be direction-preserving — it clears the READABLE
    // interest on ONE registration (EPOLL_CTL_MOD) instead of removing the fd
    // wholesale, so a writable interest on the same fd (pty master input path)
    // survives a stdout pause, and resume re-arms via MOD on the same token.
    let mut fds = [0; 2];
    unsafe { libc::pipe2(fds.as_mut_ptr(), libc::O_CLOEXEC | libc::O_NONBLOCK) };
    let r = Fd::new(fds[0], "pipe").unwrap();
    let w = Fd::new(fds[1], "pipe").unwrap();

    let mut drain: DrainState<fn(&[u8]) -> bool> =
        DrainState::new(None, None, Some(r), None, 1024, None, None, false).unwrap();
    let mut reactor = Reactor::new().unwrap();
    drain.register_with_reactor(&mut reactor).unwrap();
    let token = drain.stdout_slot.as_ref().unwrap().token.unwrap();

    // Pause stdout: registration must survive (token retained), readable
    // interest cleared.
    drain.pause_stdout(&mut reactor).unwrap();
    assert_eq!(
        drain.stdout_slot.as_ref().unwrap().token,
        Some(token),
        "pause must keep the fd registered (MOD), not del it"
    );
    // A write while paused must not wake the loop.
    w.write_slice(b"x").unwrap();
    let mut events = Vec::new();
    assert_eq!(
        reactor.wait(&mut events, 4, 50).unwrap(),
        0,
        "paused readable interest must not deliver events"
    );

    // Resume: MOD re-arms the same token; the pending byte fires again.
    assert!(drain.resume_stdout(&mut reactor).unwrap());
    let mut events = Vec::new();
    assert_eq!(reactor.wait(&mut events, 4, 100).unwrap(), 1);
    assert_eq!(events[0].token, token);
    assert!(events[0].readable);
    // Drain the pending byte.
    assert!(!drain.read_fd(true).unwrap());
    assert!(!drain.stdout_paused());
}

#[test]
fn test_pty_set_writable_arms_and_disarms_writable_interest() {
    // D4-7 / plan §5.2.2(2): the pty master (stdout slot) must be able to arm
    // a WRITABLE interest so the daemon can drain its bounded input queue on
    // POLLOUT readiness — a direction-preserving MOD on the existing
    // registration, not a del+add, so the readable interest survives.
    let master = unsafe {
        libc::open(
            c"/dev/ptmx".as_ptr(),
            libc::O_RDWR | libc::O_NOCTTY | libc::O_CLOEXEC,
        )
    };
    assert!(master >= 0);
    assert_eq!(unsafe { libc::grantpt(master) }, 0);
    assert_eq!(unsafe { libc::unlockpt(master) }, 0);
    let mut name = [0 as libc::c_char; 4096];
    assert_eq!(
        unsafe { libc::ptsname_r(master, name.as_mut_ptr(), name.len()) },
        0
    );
    let slave = unsafe {
        libc::open(
            name.as_ptr(),
            libc::O_RDWR | libc::O_NOCTTY | libc::O_CLOEXEC,
        )
    };
    assert!(slave >= 0);

    let master_fd = Fd::new(master, "test pty master").unwrap();
    let _slave_fd = Fd::new(slave, "test pty slave").unwrap();

    let mut drain: DrainState<fn(&[u8]) -> bool> =
        DrainState::new(None, None, Some(master_fd), None, 1024, None, None, true).unwrap();
    let mut reactor = Reactor::new().unwrap();
    drain.register_with_reactor(&mut reactor).unwrap();
    let token = drain.stdout_slot.as_ref().unwrap().token.unwrap();

    // pty_input_token exposes the master's registration to the daemon.
    assert_eq!(drain.pty_input_token(), Some(token));

    // Arm writable: MOD keeps the token; the master is writable immediately.
    drain.set_pty_writable(&mut reactor, true).unwrap();
    assert_eq!(drain.stdout_slot.as_ref().unwrap().token, Some(token));
    assert!(drain.stdout_slot.as_ref().unwrap().writable);
    assert!(drain.stdout_slot.as_ref().unwrap().readable);

    let mut events = Vec::new();
    assert_eq!(reactor.wait(&mut events, 4, 100).unwrap(), 1);
    assert_eq!(events[0].token, token);
    assert!(events[0].writable);

    // Disarm: readable interest survives, writable clears, no more events.
    drain.set_pty_writable(&mut reactor, false).unwrap();
    assert_eq!(drain.stdout_slot.as_ref().unwrap().token, Some(token));
    assert!(!drain.stdout_slot.as_ref().unwrap().writable);
    assert!(drain.stdout_slot.as_ref().unwrap().readable);
    let mut events = Vec::new();
    assert_eq!(
        reactor.wait(&mut events, 4, 50).unwrap(),
        0,
        "disarmed writable interest must not deliver events"
    );
}

#[test]
fn test_pty_set_writable_rejects_non_pty() {
    // D4-7: only a pty master carries an input path — a plain pipe stdout has
    // no writable route and must refuse with EINVAL, and pty_input_token must
    // be None so the daemon never routes pty-writable events to pipe jobs.
    let mut fds = [0; 2];
    unsafe { libc::pipe2(fds.as_mut_ptr(), libc::O_CLOEXEC | libc::O_NONBLOCK) };
    let r = Fd::new(fds[0], "pipe").unwrap();
    let _w = Fd::new(fds[1], "pipe").unwrap();

    let mut drain: DrainState<fn(&[u8]) -> bool> =
        DrainState::new(None, None, Some(r), None, 1024, None, None, false).unwrap();
    let mut reactor = Reactor::new().unwrap();
    drain.register_with_reactor(&mut reactor).unwrap();
    assert_eq!(drain.pty_input_token(), None);
    let err = drain.set_pty_writable(&mut reactor, true).unwrap_err();
    assert_eq!(err.raw_os_error(), Some(libc::EINVAL));
}

#[test]
fn test_managed_pty_input_token_and_writable_forward() {
    // D4-7: ManagedProcess exposes the pty input token and the writable
    // toggle; the master writable interest arms, a line is delivered to the
    // child, and the job completes.
    let mut managed = spawn_managed(
        SpawnOptions::builder(
            vec![
                "/bin/sh".to_string(),
                "-c".to_string(),
                "read line; echo got:$line".to_string(),
            ],
            SpawnBackend::Fork,
        )
        .pgroup(ProcessGroup::new(None, true))
        .pty()
        .capture_stdout()
        .timeout_ms(10_000)
        .kill_grace_ms(200)
        .cancel(CancelPolicy::Kill)
        .build()
        .unwrap(),
    )
    .unwrap();

    let mut reactor = Reactor::new().unwrap();
    managed.register_with_reactor(&mut reactor).unwrap();
    // The pty master's stdout registration is the input route.
    assert!(managed.pty_input_token().is_some());
    assert!(managed.set_pty_writable(&mut reactor, true).is_ok());
    assert!(managed.set_pty_writable(&mut reactor, false).is_ok());
    // Let the child finish its `read line`.
    assert_eq!(managed.write_input_nonblock(b"hello\r").unwrap(), Some(6));

    // Drive with the SAME reactor (a second register_with_reactor on a fresh
    // reactor would MOD an fd the fresh reactor never added).
    let mut events = Vec::new();
    let output = loop {
        if let Some(output) = managed.poll_completion(&mut reactor).unwrap() {
            break output;
        }
        let timeout = managed
            .next_deadline()
            .map(|at| {
                at.saturating_duration_since(std::time::Instant::now())
                    .as_millis()
                    .min(i32::MAX as u128) as i32
            })
            .unwrap_or(-1);
        reactor.wait(&mut events, 16, timeout).unwrap();
        for event in &events {
            managed.handle_reactor_event(&mut reactor, event).unwrap();
        }
    };
    assert!(!output.timed_out);
    let merged = String::from_utf8_lossy(&output.stdout);
    assert!(
        merged.contains("got:hello"),
        "child must read the written line, got {merged:?}"
    );
}

#[test]
fn test_spawn_output_limit_is_combined_and_errors() {
    let err = SpawnOptions::builder(
        vec![
            "/bin/sh".to_string(),
            "-c".to_string(),
            "printf 12345; printf 67890 >&2".to_string(),
        ],
        SpawnBackend::PosixSpawn,
    )
    .capture_stdout()
    .capture_stderr()
    .max_output(8)
    .build()
    .unwrap()
    .run()
    .unwrap_err();

    assert_eq!(err.raw_os_error(), Some(libc::EOVERFLOW));
}

#[test]
fn test_spawn_output_exact_limit_eof_is_not_overflow() {
    let out = SpawnOptions::builder(
        vec![
            "/bin/sh".to_string(),
            "-c".to_string(),
            "printf 12345".to_string(),
        ],
        SpawnBackend::PosixSpawn,
    )
    .capture_stdout()
    .max_output(5)
    .build()
    .unwrap()
    .run()
    .unwrap();

    assert_eq!(out.stdout, b"12345");
}

#[test]
fn test_spawn_zero_output_limit_without_output_is_not_overflow() {
    let out = SpawnOptions::builder(vec!["/bin/true".to_string()], SpawnBackend::PosixSpawn)
        .capture_stdout()
        .capture_stderr()
        .max_output(0)
        .build()
        .unwrap()
        .run()
        .unwrap();

    assert!(out.stdout.is_empty());
    assert!(out.stderr.is_empty());
}

fn drive_managed(
    mut managed: crate::spawn::ManagedProcess,
    cancel: bool,
) -> Result<crate::spawn::Output, crate::CoreError> {
    let mut reactor = Reactor::new().unwrap();
    managed.register_with_reactor(&mut reactor).unwrap();
    if cancel {
        managed.request_cancel();
    }
    let mut events = Vec::new();
    loop {
        if let Some(output) = managed.poll_completion(&mut reactor)? {
            return Ok(output);
        }
        let timeout = managed
            .next_deadline()
            .map(|at| {
                at.saturating_duration_since(std::time::Instant::now())
                    .as_millis()
                    .min(i32::MAX as u128) as i32
            })
            .unwrap_or(-1);
        reactor.wait(&mut events, 16, timeout)?;
        for event in &events {
            managed.handle_reactor_event(&mut reactor, event)?;
        }
    }
}

#[test]
fn test_spawn_managed_returns_full_output() {
    let managed = spawn_managed(
        SpawnOptions::builder(
            vec!["/bin/echo".to_string(), "managed".to_string()],
            SpawnBackend::PosixSpawn,
        )
        .capture_stdout()
        .capture_stderr()
        .build()
        .unwrap(),
    )
    .unwrap();

    let output = drive_managed(managed, false).unwrap();
    assert_eq!(output.status, Some(ExitStatus::Exited(0)));
    assert_eq!(output.stdout, b"managed\n");
    assert!(output.stderr.is_empty());
    assert!(!output.timed_out);
}

#[test]
fn test_spawn_managed_closed_stdio_fds_relocated_and_full_output() {
    // Regression: with fds 0/1/2 closed (daemonized parent), pipe2() hands
    // back the low numbers. The child setup must relocate them above stdio
    // before dup2 so the dup'd stdio survives, and the child-side pipe ends
    // must stay blocking so a large output stream is not truncated on EAGAIN.
    // Runs in a forked child so the shared stdio is never disturbed while
    // other tests run in parallel.
    let mut pipe = [0i32; 2];
    assert_eq!(unsafe { libc::pipe(pipe.as_mut_ptr()) }, 0);
    let pid = unsafe { libc::fork() };
    assert!(pid >= 0, "fork failed");
    if pid == 0 {
        unsafe {
            libc::close(pipe[0]);
        }
        let _ = unsafe { libc::close(0) };
        let _ = unsafe { libc::close(1) };
        let _ = unsafe { libc::close(2) };
        let result = spawn_managed(
            SpawnOptions::builder(
                vec![
                    "/bin/sh".to_string(),
                    "-c".to_string(),
                    "i=0; while [ $i -lt 4000 ]; do echo chunk-$i; i=$((i+1)); done".to_string(),
                ],
                SpawnBackend::Fork,
            )
            .capture_stdout()
            .build()
            .unwrap(),
        )
        .and_then(|managed| drive_managed(managed, false));
        let ok = matches!(result, Ok(output)
            if output.status == Some(ExitStatus::Exited(0))
                && output.stdout.starts_with(b"chunk-0\n")
                && output.stdout.ends_with(b"chunk-3999\n"));
        let buf = [ok as i8, 0];
        unsafe {
            libc::write(pipe[1], buf.as_ptr().cast(), buf.len());
            libc::close(pipe[1]);
            libc::_exit(if ok { 0 } else { 1 });
        }
    }
    unsafe {
        libc::close(pipe[1]);
    }
    let mut got = [0i8; 2];
    let mut off = 0usize;
    while off < got.len() {
        let n = unsafe { libc::read(pipe[0], got[off..].as_mut_ptr().cast(), got.len() - off) };
        if n <= 0 {
            break;
        }
        off += n as usize;
    }
    unsafe {
        libc::close(pipe[0]);
    }
    let mut wait_status = 0;
    unsafe {
        libc::waitpid(pid, &mut wait_status, 0);
    }
    assert_eq!(got[0], 1, "child spawn failed or stdout truncated");
}

#[test]
fn test_spawn_managed_exact_limit_is_not_overflow() {
    let managed = spawn_managed(
        SpawnOptions::builder(
            vec!["/bin/echo".to_string(), "1234".to_string()],
            SpawnBackend::PosixSpawn,
        )
        .capture_stdout()
        .max_output(5)
        .build()
        .unwrap(),
    )
    .unwrap();

    let output = drive_managed(managed, false).unwrap();
    assert_eq!(output.stdout, b"1234\n");
}

#[test]
fn test_spawn_managed_reports_output_overflow() {
    let managed = spawn_managed(
        SpawnOptions::builder(
            vec!["/bin/echo".to_string(), "12345".to_string()],
            SpawnBackend::PosixSpawn,
        )
        .capture_stdout()
        .max_output(5)
        .build()
        .unwrap(),
    )
    .unwrap();

    let err = drive_managed(managed, false).unwrap_err();
    assert_eq!(err.raw_os_error(), Some(libc::EOVERFLOW));
}

#[test]
fn test_spawn_managed_timeout_kills_own_process_group() {
    let managed = spawn_managed(
        SpawnOptions::builder(
            vec!["/bin/sleep".to_string(), "10".to_string()],
            SpawnBackend::PosixSpawn,
        )
        .pgroup(ProcessGroup::new(Some(0), false))
        .timeout_ms(20)
        .kill_grace_ms(20)
        .cancel(CancelPolicy::Graceful)
        .build()
        .unwrap(),
    )
    .unwrap();

    let output = drive_managed(managed, false).unwrap();
    assert!(output.timed_out);
    assert!(matches!(output.status, Some(ExitStatus::Signaled(_))));
}

#[test]
fn test_spawn_managed_explicit_cancel_reaps_child() {
    let managed = spawn_managed(
        SpawnOptions::builder(
            vec!["/bin/sleep".to_string(), "10".to_string()],
            SpawnBackend::PosixSpawn,
        )
        .pgroup(ProcessGroup::new(Some(0), false))
        .cancel(CancelPolicy::Kill)
        .build()
        .unwrap(),
    )
    .unwrap();

    let output = drive_managed(managed, true).unwrap();
    assert!(!output.timed_out);
    assert!(matches!(output.status, Some(ExitStatus::Signaled(_))));
}

#[test]
fn test_managed_pid_available_after_completion() {
    let managed = spawn_managed(
        SpawnOptions::builder(
            vec!["/bin/echo".to_string(), "managed".to_string()],
            SpawnBackend::PosixSpawn,
        )
        .capture_stdout()
        .build()
        .unwrap(),
    )
    .unwrap();
    let pid = managed.pid();
    assert!(pid > 0);

    let output = drive_managed(managed, false).unwrap();
    // The pid must remain queryable after the running handle was consumed.
    assert_eq!(output.pid, pid);
}

#[test]
fn test_managed_register_twice_is_idempotent() {
    let mut managed = spawn_managed(
        SpawnOptions::builder(
            vec!["/bin/echo".to_string(), "managed".to_string()],
            SpawnBackend::PosixSpawn,
        )
        .capture_stdout()
        .build()
        .unwrap(),
    )
    .unwrap();

    let mut reactor = Reactor::new().unwrap();
    // A second registration must not fail or destroy the registered streams.
    managed.register_with_reactor(&mut reactor).unwrap();
    managed.register_with_reactor(&mut reactor).unwrap();

    let mut events = Vec::new();
    loop {
        if let Some(output) = managed.poll_completion(&mut reactor).unwrap() {
            assert_eq!(output.stdout, b"managed\n");
            return;
        }
        let timeout = managed
            .next_deadline()
            .map(|at| {
                at.saturating_duration_since(std::time::Instant::now())
                    .as_millis()
                    .min(i32::MAX as u128) as i32
            })
            .unwrap_or(-1);
        reactor.wait(&mut events, 16, timeout).unwrap();
        for event in &events {
            managed.handle_reactor_event(&mut reactor, event).unwrap();
        }
    }
}

#[test]
fn test_spawn_chunk_sink_forwards_chunks_streaming() {
    let received_stdout = Arc::new(Mutex::new(Vec::<u8>::new()));
    let received_stderr = Arc::new(Mutex::new(Vec::<u8>::new()));
    let r_out = Arc::clone(&received_stdout);
    let r_err = Arc::clone(&received_stderr);
    let out = SpawnOptions::builder(
        vec![
            "/bin/sh".to_string(),
            "-c".to_string(),
            "printf hello; printf world; printf err >&2".to_string(),
        ],
        SpawnBackend::PosixSpawn,
    )
    .capture_stdout()
    .capture_stderr()
    .chunk_sink(move |is_stdout, bytes| {
        if is_stdout {
            r_out.lock().unwrap().extend_from_slice(bytes);
        } else {
            r_err.lock().unwrap().extend_from_slice(bytes);
        }
        SinkResult::Accept
    })
    .build()
    .unwrap()
    .run()
    .unwrap();

    assert_eq!(&*received_stdout.lock().unwrap(), b"helloworld");
    assert_eq!(&*received_stderr.lock().unwrap(), b"err");
    // Streaming mode: nothing accumulates into the completion buffers.
    assert!(out.stdout.is_empty());
    assert!(out.stderr.is_empty());
    // Every chunk was delivered; nothing held for a terminal flush.
    assert!(out.stdout_pending.is_none());
    assert!(out.stderr_pending.is_none());
    assert_eq!(out.status, Some(ExitStatus::Exited(0)));
}

#[test]
fn test_spawn_chunk_sink_pause_resume_lossless_with_consumer() {
    const CAP: usize = 1; // accept one chunk, then pause until the consumer drains
    let outstanding = Arc::new(AtomicUsize::new(0));
    let received = Arc::new(Mutex::new(Vec::<u8>::new()));
    let stop = Arc::new(AtomicBool::new(false));

    let sink_out = Arc::clone(&outstanding);
    let sink_recv = Arc::clone(&received);
    let sink = move |_is_stdout, bytes: &[u8]| {
        if sink_out.load(Ordering::SeqCst) >= CAP {
            return SinkResult::Pause;
        }
        sink_out.fetch_add(1, Ordering::SeqCst);
        sink_recv.lock().unwrap().extend_from_slice(bytes);
        SinkResult::Accept
    };

    let consumer_out = Arc::clone(&outstanding);
    let consumer_stop = Arc::clone(&stop);
    let consumer = std::thread::spawn(move || {
        while !consumer_stop.load(Ordering::SeqCst) {
            consumer_out.store(0, Ordering::SeqCst);
            std::thread::sleep(std::time::Duration::from_millis(1));
        }
    });

    let out = SpawnOptions::builder(
        vec![
            "/bin/sh".to_string(),
            "-c".to_string(),
            "dd if=/dev/zero bs=1 count=300000 2>/dev/null | tr '\\000' x".to_string(),
        ],
        SpawnBackend::PosixSpawn,
    )
    .capture_stdout()
    .chunk_sink(sink)
    .build()
    .unwrap()
    .run()
    .unwrap();

    // Flush any held chunk the same way the daemon does before a terminal frame.
    if let Some(p) = out.stdout_pending {
        let pending = p;
        loop {
            if outstanding.load(Ordering::SeqCst) < CAP {
                outstanding.fetch_add(1, Ordering::SeqCst);
                received.lock().unwrap().extend_from_slice(&pending);
                break;
            }
            std::thread::sleep(std::time::Duration::from_millis(1));
        }
    }
    stop.store(true, Ordering::SeqCst);
    consumer.join().unwrap();

    // Lossless across pause/resume cycles: every byte, in order, exactly once.
    let got = received.lock().unwrap();
    assert_eq!(got.len(), 300000);
    assert!(got.iter().all(|&b| b == b'x'));
    // Streaming mode: nothing accumulated into the completion buffers.
    assert!(out.stdout.is_empty());
}

#[test]
fn test_spawn_chunk_sink_pause_at_end_pending_flushed() {
    // A sink that never accepts: the first read chunk is held as pending.
    let received = Arc::new(Mutex::new(Vec::<u8>::new()));
    let out = SpawnOptions::builder(
        vec![
            "/bin/sh".to_string(),
            "-c".to_string(),
            "printf helddata".to_string(),
        ],
        SpawnBackend::PosixSpawn,
    )
    .capture_stdout()
    .chunk_sink(move |_is_stdout, _bytes| SinkResult::Pause)
    .build()
    .unwrap()
    .run()
    .unwrap();

    assert!(received.lock().unwrap().is_empty());
    // Nothing was accepted; the held chunk is returned for the terminal flush.
    assert_eq!(out.stdout_pending.as_deref(), Some(b"helddata".as_slice()));
    assert!(out.stdout.is_empty());
    assert_eq!(out.status, Some(ExitStatus::Exited(0)));
}

#[test]
fn test_managed_chunk_sink_pause_resume_lossless() {
    const CAP: usize = 1;
    let outstanding = Arc::new(AtomicUsize::new(0));
    let received = Arc::new(Mutex::new(Vec::<u8>::new()));
    let stop = Arc::new(AtomicBool::new(false));

    let sink_out = Arc::clone(&outstanding);
    let sink_recv = Arc::clone(&received);
    let sink = move |_is_stdout, bytes: &[u8]| {
        if sink_out.load(Ordering::SeqCst) >= CAP {
            return SinkResult::Pause;
        }
        sink_out.fetch_add(1, Ordering::SeqCst);
        sink_recv.lock().unwrap().extend_from_slice(bytes);
        SinkResult::Accept
    };

    let consumer_out = Arc::clone(&outstanding);
    let consumer_stop = Arc::clone(&stop);
    let consumer = std::thread::spawn(move || {
        while !consumer_stop.load(Ordering::SeqCst) {
            consumer_out.store(0, Ordering::SeqCst);
            std::thread::sleep(std::time::Duration::from_millis(1));
        }
    });

    let mut managed = spawn_managed(
        SpawnOptions::builder(
            vec![
                "/bin/sh".to_string(),
                "-c".to_string(),
                "dd if=/dev/zero bs=1 count=300000 2>/dev/null | tr '\\000' x".to_string(),
            ],
            SpawnBackend::PosixSpawn,
        )
        .capture_stdout()
        .chunk_sink(sink)
        .build()
        .unwrap(),
    )
    .unwrap();

    let mut reactor = Reactor::new().unwrap();
    managed.register_with_reactor(&mut reactor).unwrap();
    let mut events = Vec::new();
    let mut output = None;
    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(20);
    while std::time::Instant::now() < deadline {
        if let Some(out) = managed.poll_completion(&mut reactor).unwrap() {
            output = Some(out);
            break;
        }
        // The daemon drains its queue, then resumes a paused stream.
        if managed.stdout_paused() {
            let _ = managed.resume_stdout(&mut reactor);
        }
        if managed.stderr_paused() {
            let _ = managed.resume_stderr(&mut reactor);
        }
        let timeout = managed
            .next_deadline()
            .map(|at| {
                at.saturating_duration_since(std::time::Instant::now())
                    .as_millis()
                    .min(i32::MAX as u128) as i32
            })
            .unwrap_or(-1);
        reactor.wait(&mut events, 16, timeout).unwrap();
        for event in &events {
            managed.handle_reactor_event(&mut reactor, event).unwrap();
        }
    }
    // Stop the consumer regardless of the outcome so the join cannot hang.
    stop.store(true, Ordering::SeqCst);
    consumer.join().unwrap();

    let out = output.expect("managed streaming child completed");
    // Flush any held chunk the way the daemon does before the terminal frame.
    if let Some(p) = out.stdout_pending {
        let pending = p;
        loop {
            if outstanding.load(Ordering::SeqCst) < CAP {
                outstanding.fetch_add(1, Ordering::SeqCst);
                received.lock().unwrap().extend_from_slice(&pending);
                break;
            }
            std::thread::sleep(std::time::Duration::from_millis(1));
        }
    }
    let got = received.lock().unwrap();
    assert_eq!(got.len(), 300000);
    assert!(got.iter().all(|&b| b == b'x'));
}

#[test]
fn test_kill_rejects_zero_pid() {
    let p = Process::new(0);
    assert!(p.kill(libc::SIGTERM).is_err());
}

#[test]
fn test_kill_group_rejects_nonpositive_pgid() {
    let p = Process::new(1);
    assert!(p.kill_group(0, libc::SIGTERM).is_err());
    assert!(p.kill_group(-1, libc::SIGTERM).is_err());
    assert!(p.kill_group(99999999, libc::SIGTERM).is_ok());
}

#[test]
fn test_fork_rejects_isolated_with_custom_leader() {
    let opts = SpawnOptions::builder(vec!["/bin/true".to_string()], SpawnBackend::Fork)
        .pgroup(ProcessGroup::new(Some(12345), true))
        .build()
        .unwrap();

    let err = match spawn_start(opts) {
        Ok(_) => panic!("expected isolated+custom-leader to fail"),
        Err(err) => err,
    };
    assert_eq!(err.raw_os_error(), Some(libc::EINVAL));
}

#[test]
fn test_managed_timeout_with_overflow_returns_partial() {
    // A wedged pipe (a descendant holding the write end) plus an overflow must
    // mirror blocking spawn's N4 behavior: Ok(partial, timed_out), not
    // EOVERFLOW, because the drain never completed.
    let managed = spawn_managed(
        SpawnOptions::builder(
            vec![
                "/bin/sh".to_string(),
                "-c".to_string(),
                "printf 1234567890; sleep 30 & exec sleep 30".to_string(),
            ],
            SpawnBackend::PosixSpawn,
        )
        .capture_stdout()
        .max_output(5)
        .timeout_ms(100)
        .kill_grace_ms(50)
        .cancel(CancelPolicy::Kill)
        .build()
        .unwrap(),
    )
    .unwrap();

    let output = drive_managed(managed, false).unwrap();
    assert!(output.timed_out);
    assert!(matches!(output.status, Some(ExitStatus::Signaled(_))));
}

#[test]
fn test_writer_state_epipe() {
    use crate::io::writer::WriterState;

    let mut fds = [0; 2];
    unsafe { libc::pipe(fds.as_mut_ptr()) };
    let r = Fd::new(fds[0], "pipe").unwrap();
    let w = Fd::new(fds[1], "pipe").unwrap();

    let mut writer = WriterState::new(Some(vec![0u8; 1024 * 1024].into_boxed_slice()));

    // Close read end to trigger EPIPE on next write
    drop(r);

    // Some kernels might not trigger EPIPE on the first write if the pipe buffer has space,
    // but with enough data it will fail.
    let mut last_res = Ok(false);
    for _ in 0..100 {
        last_res = writer.write_to_fd(&w);
        if last_res.is_err() || (last_res.is_ok() && writer.buf.is_none()) {
            break;
        }
    }

    // EPIPE should be handled as "done" (Ok(true))
    assert!(last_res.is_ok());
    assert!(last_res.unwrap());
    assert!(writer.buf.is_none());
}

#[test]
fn test_path_existence() {
    let temp_file = std::env::temp_dir().join("coreshift_test_path");
    let path_str = temp_file.to_str().unwrap();

    std::fs::write(&temp_file, "test").unwrap();
    assert!(path_exists(path_str));
    assert!(path_lstat_exists(path_str));

    std::fs::remove_file(&temp_file).unwrap();
    assert!(!path_exists(path_str));
    assert!(!path_lstat_exists(path_str));
}

#[test]
fn test_path_uid_temp_file() {
    let path = std::env::temp_dir().join(format!(
        "coreshift_test_uid_{}_{}",
        std::process::id(),
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_nanos()
    ));
    std::fs::write(&path, b"uid").unwrap();

    let uid = path_uid(&path).unwrap();
    assert_eq!(uid, unsafe { libc::geteuid() });

    remove_file(&path).unwrap();
}

#[test]
fn test_path_stat_reports_identity_fields() {
    let path = std::env::temp_dir().join(format!(
        "coreshift_test_path_stat_{}_{}",
        std::process::id(),
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_nanos()
    ));
    std::fs::write(&path, b"identity").unwrap();

    let stat = path_stat(&path).unwrap();
    assert_eq!(stat.uid, unsafe { libc::geteuid() });
    assert!(stat.inode > 0);

    remove_file(&path).unwrap();
}

#[test]
fn test_path_stat_follows_symlink_and_lstat_reports_link() {
    let dir = std::env::temp_dir().join(format!(
        "coreshift_test_symlink_stat_{}_{}",
        std::process::id(),
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_nanos()
    ));
    std::fs::create_dir(&dir).unwrap();
    let target = dir.join("target");
    let link = dir.join("link");
    std::fs::write(&target, b"identity").unwrap();
    std::os::unix::fs::symlink(&target, &link).unwrap();

    let target_stat = path_stat(&target).unwrap();
    let followed = path_stat(&link).unwrap();
    let link_stat = path_lstat(&link).unwrap();

    assert_eq!(followed.inode, target_stat.inode);
    assert_ne!(link_stat.inode, target_stat.inode);

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

#[test]
fn test_clock_ticks_per_second_checked_result() {
    let ticks = clock_ticks_per_second().unwrap();
    assert!(ticks > 0);
}

#[test]
fn test_proc_uid_current_process() {
    let uid = uid(std::process::id() as i32).unwrap();
    assert_eq!(uid, unsafe { libc::geteuid() });
}

#[test]
fn test_proc_uid_at_uses_explicit_root() {
    let proc_root = std::env::temp_dir().join(format!(
        "coreshift_test_proc_uid_root_{}_{}",
        std::process::id(),
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_nanos()
    ));
    let pid_dir = proc_root.join("54321");
    std::fs::create_dir_all(&pid_dir).unwrap();

    let uid = uid_at(&proc_root, 54321).unwrap();
    assert_eq!(uid, unsafe { libc::geteuid() });

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

#[test]
fn test_proc_stat_at_uses_explicit_root() {
    let proc_root = std::env::temp_dir().join(format!(
        "coreshift_test_proc_stat_root_{}_{}",
        std::process::id(),
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_nanos()
    ));
    let pid_dir = proc_root.join("54321");
    std::fs::create_dir_all(&pid_dir).unwrap();

    let stat = stat_at(&proc_root, 54321).unwrap();
    assert_eq!(stat.uid, unsafe { libc::geteuid() });
    assert!(stat.inode > 0);

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

#[test]
fn test_proc_uid_invalid_pid_returns_error() {
    let err = uid(999_999).unwrap_err();
    assert_eq!(err.raw_os_error(), Some(libc::ENOENT));
}

#[test]
fn test_path_uid_missing_path_returns_error() {
    let path = std::env::temp_dir().join(format!(
        "coreshift_missing_uid_{}_{}",
        std::process::id(),
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_nanos()
    ));
    let err = path_uid(&path).unwrap_err();
    assert_eq!(err.raw_os_error(), Some(libc::ENOENT));
}

#[test]
fn test_install_shutdown_flag_compiles_and_sets_helper_state() {
    let _lock = SIGNAL_TEST_LOCK.lock().unwrap();
    TEST_SHUTDOWN_FLAG_A.store(false, Ordering::Release);
    install_shutdown_flag(&TEST_SHUTDOWN_FLAG_A).unwrap();
    assert!(!shutdown_requested(&TEST_SHUTDOWN_FLAG_A));

    TEST_SHUTDOWN_FLAG_A.store(true, Ordering::Release);
    assert!(shutdown_requested(&TEST_SHUTDOWN_FLAG_A));
    TEST_SHUTDOWN_FLAG_A.store(false, Ordering::Release);
}

#[test]
fn test_install_shutdown_flag_repeated_install_does_not_panic() {
    let _lock = SIGNAL_TEST_LOCK.lock().unwrap();
    TEST_SHUTDOWN_FLAG_A.store(false, Ordering::Release);
    TEST_SHUTDOWN_FLAG_B.store(false, Ordering::Release);

    install_shutdown_flag(&TEST_SHUTDOWN_FLAG_A).unwrap();
    install_shutdown_flag(&TEST_SHUTDOWN_FLAG_B).unwrap();

    assert!(!shutdown_requested(&TEST_SHUTDOWN_FLAG_A));
    assert!(!shutdown_requested(&TEST_SHUTDOWN_FLAG_B));
}

#[test]
fn test_spawn_resets_inherited_ignored_signals() {
    let _lock = SIGNAL_TEST_LOCK.lock().unwrap();
    // Simulate a backgrounded launch (POSIX: `cmd &` sets SIGHUP/SIGINT/SIGQUIT
    // to SIG_IGN in the child). A spawned child inheriting these stays immune
    // to terminal control signals forever.
    let saved = unsafe {
        [
            libc::signal(libc::SIGHUP, libc::SIG_IGN),
            libc::signal(libc::SIGINT, libc::SIG_IGN),
            libc::signal(libc::SIGQUIT, libc::SIG_IGN),
        ]
    };
    assert_ne!(saved[0], libc::SIG_ERR);
    assert_ne!(saved[1], libc::SIG_ERR);
    assert_ne!(saved[2], libc::SIG_ERR);

    for backend in [SpawnBackend::Clone3Pidfd, SpawnBackend::Fork] {
        // If SIGINT were still SIG_IGN, `kill -INT $$` would be a no-op and
        // the shell would print SURVIVED. With the reset it dies on SIGINT.
        let managed = spawn_managed(
            SpawnOptions::builder(
                vec![
                    "/bin/sh".to_string(),
                    "-c".to_string(),
                    "kill -INT $$; echo SURVIVED".to_string(),
                ],
                backend,
            )
            .capture_stdout()
            .timeout_ms(5000)
            .cancel(CancelPolicy::Graceful)
            .build()
            .unwrap(),
        )
        .unwrap();
        let output = drive_managed(managed, false).unwrap();
        let stdout = String::from_utf8_lossy(&output.stdout).to_string();
        assert!(
            !stdout.contains("SURVIVED"),
            "{backend:?}: child inherited SIG_IGN for SIGINT (SURVIVED printed)"
        );
        assert!(
            matches!(output.status, Some(ExitStatus::Signaled(libc::SIGINT))),
            "{backend:?}: expected child killed by SIGINT, got {:?}",
            output.status
        );
    }

    unsafe {
        libc::signal(libc::SIGHUP, saved[0]);
        libc::signal(libc::SIGINT, saved[1]);
        libc::signal(libc::SIGQUIT, saved[2]);
    }
}

#[test]
fn test_shutdown_flag_guard_restores_previous_handler() {
    let _lock = SIGNAL_TEST_LOCK.lock().unwrap();
    let mut ignore_action: libc::sigaction = unsafe { std::mem::zeroed() };
    let mut old_action: libc::sigaction = unsafe { std::mem::zeroed() };
    ignore_action.sa_sigaction = libc::SIG_IGN;
    unsafe { libc::sigemptyset(&mut ignore_action.sa_mask) };

    let ret = unsafe { libc::sigaction(libc::SIGTERM, &ignore_action, &mut old_action) };
    assert_eq!(ret, 0);

    {
        let _guard = install_shutdown_flag_guard(&TEST_SHUTDOWN_FLAG_A).unwrap();
    }

    let mut current_action: libc::sigaction = unsafe { std::mem::zeroed() };
    let ret = unsafe { libc::sigaction(libc::SIGTERM, std::ptr::null(), &mut current_action) };
    assert_eq!(ret, 0);
    assert_eq!(current_action.sa_sigaction, libc::SIG_IGN);

    let ret = unsafe { libc::sigaction(libc::SIGTERM, &old_action, std::ptr::null_mut()) };
    assert_eq!(ret, 0);
}

#[test]
fn test_reactor_setup_signalfd_restores_previous_mask_on_drop() {
    let mut before: libc::sigset_t = unsafe { std::mem::zeroed() };
    let ret = unsafe { libc::pthread_sigmask(libc::SIG_SETMASK, std::ptr::null(), &mut before) };
    assert_eq!(ret, 0);
    let was_blocked = unsafe { libc::sigismember(&before, libc::SIGCHLD) };

    {
        let mut reactor = Reactor::new().unwrap();
        reactor.setup_signalfd().unwrap();
        let mut during: libc::sigset_t = unsafe { std::mem::zeroed() };
        let ret =
            unsafe { libc::pthread_sigmask(libc::SIG_SETMASK, std::ptr::null(), &mut during) };
        assert_eq!(ret, 0);
        assert_eq!(unsafe { libc::sigismember(&during, libc::SIGCHLD) }, 1);
    }

    let mut after: libc::sigset_t = unsafe { std::mem::zeroed() };
    let ret = unsafe { libc::pthread_sigmask(libc::SIG_SETMASK, std::ptr::null(), &mut after) };
    assert_eq!(ret, 0);
    assert_eq!(
        unsafe { libc::sigismember(&after, libc::SIGCHLD) },
        was_blocked
    );
}

#[test]
fn test_reactor_setup_signalfd_rejects_repeated_setup() {
    let mut reactor = Reactor::new().unwrap();
    reactor.setup_signalfd().unwrap();

    let err = reactor.setup_signalfd().unwrap_err();
    assert_eq!(err.raw_os_error(), Some(libc::EINVAL));
}

#[test]
fn test_reactor_signalfd_drop_on_other_thread_does_not_clobber_mask() {
    // Regression: dropping a reactor whose SIGCHLD mask was set on another
    // thread must not restore that thread's `previous_mask` onto the *current*
    // (dropping) thread. `Drop` restores only when it runs on the setup
    // thread; a cross-thread drop leaves the dropping thread's mask untouched.
    let mut reactor = Reactor::new().unwrap();
    reactor.setup_signalfd().unwrap();

    let result = std::sync::Arc::new(std::sync::Mutex::new(None));
    let result_clone = std::sync::Arc::clone(&result);
    let handle = std::thread::spawn(move || {
        // Give this thread a sentinel mask bit so a clobber is detectable.
        let mut sentinel: libc::sigset_t = unsafe { std::mem::zeroed() };
        unsafe { libc::sigemptyset(&mut sentinel) };
        unsafe { libc::sigaddset(&mut sentinel, libc::SIGUSR2) };
        let ret =
            unsafe { libc::pthread_sigmask(libc::SIG_SETMASK, &sentinel, std::ptr::null_mut()) };
        assert_eq!(ret, 0);

        drop(reactor);

        // If the buggy drop applied the setup thread's previous mask here,
        // SIGUSR2 would be unblocked. With the fix nothing is restored.
        let mut current: libc::sigset_t = unsafe { std::mem::zeroed() };
        let ret =
            unsafe { libc::pthread_sigmask(libc::SIG_SETMASK, std::ptr::null(), &mut current) };
        assert_eq!(ret, 0);
        let sentinel_kept = unsafe { libc::sigismember(&current, libc::SIGUSR2) };
        *result_clone.lock().unwrap() = Some(sentinel_kept);
    });
    handle.join().unwrap();

    assert_eq!(result.lock().unwrap().take(), Some(1));
}

#[test]
fn test_readahead_small_temp_file() {
    with_temp_readahead_file(|file, _| {
        assert_readahead_result(readahead(file, 0, 16));
    });
}

#[test]
fn test_readahead_zero_length() {
    with_temp_readahead_file(|file, _| {
        assert_readahead_result(readahead(file, 0, 0));
    });
}

#[test]
fn test_readahead_offset_beyond_eof() {
    with_temp_readahead_file(|file, _| {
        assert_readahead_result(readahead(file, 1 << 20, 16));
    });
}

#[test]
fn test_readahead_invalid_fd() {
    let result = readahead(RawFdRef(-1), 0, 16);
    match result {
        Err(err) if err.raw_os_error() == Some(libc::EBADF) => {}
        Err(err) if err.raw_os_error() == Some(libc::ENOSYS) => {
            eprintln!("skipping readahead test: unsupported on this target");
        }
        Err(err) => panic!("expected EBADF from invalid fd, got: {err}"),
        Ok(()) => panic!("expected invalid fd to fail"),
    }
}

#[test]
fn test_mmap_madvise_offset_zero() {
    with_temp_readahead_file(|file, _| {
        let result = mmap_madvise(file, 0, 16, false);
        if let Err(err) = &result
            && err.raw_os_error() == Some(libc::ENOSYS)
        {
            eprintln!("skipping mmap_madvise test: unsupported on this target");
            return;
        }
        result.unwrap();
    });
}

#[test]
fn test_mmap_madvise_rejects_unaligned_offset() {
    with_temp_readahead_file(|file, _| {
        let result = mmap_madvise(file, 1, 16, false);
        assert_eq!(result.unwrap_err().raw_os_error(), Some(libc::EINVAL));
    });
}