openjd-sessions 0.5.5

// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
// Copyright by contributors to this project.
// SPDX-License-Identifier: (Apache-2.0 OR MIT)

//! Async subprocess execution with real-time message streaming.

use std::borrow::Cow;
use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::time::Duration;

use tokio::io::{AsyncBufReadExt, BufReader};
use tokio::process::Command;
use tokio::sync::mpsc;
use tokio_util::sync::CancellationToken;

use crate::action::ActionMessage;
use crate::action::ActionState;
use crate::action_filter::{ActionFilter, ActionMessageKind, ActionMessageValue};
use crate::error::SessionError;
use crate::logging::LogContent;
use crate::runner::CancelMethod;
use crate::session_log;
use crate::session_user::SessionUser;
use std::sync::Arc;

/// Grace time to wait for `c.wait()` to reap the child when the stdout
/// read loop exited through a non-kill path (natural EOF, read error) or
/// after a `NotifyThenTerminate` cancellation where the process may still
/// be winding down gracefully in response to the notify signal. Used
/// when no terminate has been issued yet from inside the stdout-read
/// loop.
const STDOUT_GRACE_TIME: Duration = Duration::from_secs(5);

/// Grace time to wait for `c.wait()` to reap the child after we have
/// already issued `send_terminate` from inside the stdout-read loop
/// (timeout fired, urgent cancel with `time_limit=0`, or
/// `CancelMethod::Terminate`).
///
/// `send_terminate` resolves to SIGKILL on Unix and `TerminateProcess`
/// (via `kill_process_tree`) on Windows. In these cases the process
/// tree was killed at least `STDOUT_DRAIN_AFTER_KILL` (1s) ago — by
/// the time the drain deadline fires and we get here, `c.wait()`
/// should reap the already-dead child almost immediately. Two seconds
/// is a generous bound for runtime scheduling delay (especially on
/// loaded Windows CI) while still being much tighter than the 5s we
/// allow on graceful-exit paths.
///
/// A shorter bound here means the three "timeout should have fired
/// quickly" integration tests in `tests/test_session.rs` don't pick up
/// 3-4s of dead time on every run, which in turn means their 10s
/// assertion holds with comfortable margin on slow CI.
const STDOUT_GRACE_TIME_POST_TERMINATE: Duration = Duration::from_secs(2);

/// Grace time to drain stdout after sending a kill signal.
const STDOUT_DRAIN_AFTER_KILL: Duration = Duration::from_secs(1);

/// Maximum line length for stdout reading.
pub(crate) const LOG_LINE_MAX_LENGTH: usize = 64 * 1024;

/// Truncate a line to at most `LOG_LINE_MAX_LENGTH` bytes on a valid UTF-8 char boundary.
pub(crate) fn truncate_line(line: &str) -> &str {
    if line.len() > LOG_LINE_MAX_LENGTH {
        &line[..line.floor_char_boundary(LOG_LINE_MAX_LENGTH)]
    } else {
        line
    }
}

/// Lowercase hex digits, indexed by nibble value.
const HEX_DIGITS: [u8; 16] = *b"0123456789abcdef";

/// Decode subprocess output as UTF-8, escaping every byte that is not valid
/// UTF-8 as `\xNN` with lowercase hex.
///
/// This mirrors CPython's `bytes.decode("utf-8", errors="backslashreplace")`,
/// which `openjd-sessions-for-python` uses when reading subprocess output. A
/// subprocess can emit bytes that are not valid UTF-8, for example a Windows
/// DCC application writing its output in the system code page, such as Unreal
/// Engine emitting the cp1252 em dash `0x97`. Escaping those bytes rather than
/// replacing them with U+FFFD preserves the original byte values in the session
/// log, which helps identify the code page the subprocess is emitting.
///
/// Valid UTF-8, including multi-byte sequences, passes through unmodified, and a
/// borrowed string is returned without allocating when the whole input is
/// already valid.
pub(crate) fn decode_backslashreplace(bytes: &[u8]) -> Cow<'_, str> {
    // Fast path: the overwhelmingly common case is fully valid UTF-8, which
    // needs no allocation.
    if let Ok(valid) = std::str::from_utf8(bytes) {
        return Cow::Borrowed(valid);
    }

    // `utf8_chunks` splits the input into (valid UTF-8 prefix, invalid byte
    // sequence) pairs, so the valid parts need no re-validation and the invalid
    // sequences are delimited exactly as UTF-8 validation defines them.
    // CPython escapes every byte of an undecodable sequence individually, so a
    // 3-byte invalid sequence becomes three `\xNN` escapes.
    let mut out = String::with_capacity(bytes.len());
    for chunk in bytes.utf8_chunks() {
        out.push_str(chunk.valid());
        for &byte in chunk.invalid() {
            out.push('\\');
            out.push('x');
            // Both indices are nibbles, so they are always < 16.
            out.push(HEX_DIGITS[(byte >> 4) as usize] as char);
            out.push(HEX_DIGITS[(byte & 0x0f) as usize] as char);
        }
    }
    Cow::Owned(out)
}

/// Result of running a subprocess action.
#[derive(Debug)]
pub struct SubprocessResult {
    pub state: ActionState,
    pub exit_code: Option<i32>,
    pub stdout: String,
}

/// Configuration for running a subprocess.
pub struct SubprocessConfig {
    pub args: Vec<String>,
    pub env_vars: HashMap<String, Option<String>>,
    pub working_dir: Option<PathBuf>,
    pub timeout: Option<Duration>,
    pub user: Option<Arc<dyn SessionUser>>,
    pub cancel_method: CancelMethod,
    pub cancel_request_rx: Option<tokio::sync::watch::Receiver<Option<Duration>>>,
    /// Whether to accumulate all stdout into `SubprocessResult.stdout`.
    /// Intended for debugging only — production callers should leave this
    /// `false` and observe output through the real-time callback.
    /// Default is `false` — lines are still streamed through the filter and
    /// callback in real time, but the collected string stays empty.
    pub debug_collect_stdout: bool,
}

// ---------------------------------------------------------------------------
// Platform-specific signal / process-group helpers
// ---------------------------------------------------------------------------

#[cfg(unix)]
mod platform {
    use super::*;

    /// Send SIGTERM to the process group.
    pub fn notify_process_group(pgid: i32) -> Result<(), std::io::Error> {
        nix::sys::signal::killpg(
            nix::unistd::Pid::from_raw(pgid),
            nix::sys::signal::Signal::SIGTERM,
        )
        .map_err(std::io::Error::other)
    }

    /// Send SIGKILL to the process group.
    pub fn terminate_process_group(pgid: i32) -> Result<(), std::io::Error> {
        nix::sys::signal::killpg(
            nix::unistd::Pid::from_raw(pgid),
            nix::sys::signal::Signal::SIGKILL,
        )
        .map_err(std::io::Error::other)
    }

    /// Send SIGKILL to the process group.
    pub fn send_terminate(pid: i32) {
        let _ = terminate_process_group(pid);
    }

    /// Send SIGTERM to the process group.
    pub fn send_notify(pid: i32) {
        let _ = notify_process_group(pid);
    }

    /// Spawn a delayed SIGKILL after a grace period.
    pub fn spawn_delayed_terminate(pid: i32, delay: Duration) {
        tokio::spawn(async move {
            tokio::time::sleep(delay).await;
            let _ = terminate_process_group(pid);
        });
    }

    /// Configure the Command for POSIX: setsid + dup2 stderr→stdout via pre_exec.
    ///
    /// Returns `None` — on POSIX the merge happens in the child via dup2,
    /// so the caller reads from `child.stdout` as normal.
    ///
    /// # Safety
    /// Calls `pre_exec` which runs in the forked child before exec.
    pub unsafe fn configure_command(
        cmd: &mut Command,
        use_setsid: bool,
    ) -> Option<Box<dyn tokio::io::AsyncRead + Unpin + Send>> {
        cmd.pre_exec(move || {
            // Redirect stderr to stdout so output ordering is preserved
            if nix::libc::dup2(1, 2) == -1 {
                return Err(std::io::Error::last_os_error());
            }
            if use_setsid {
                nix::libc::setsid();
            }
            Ok(())
        });
        None
    }
}

#[cfg(windows)]
mod platform {
    use super::*;

    use windows::Win32::Foundation::{CloseHandle, FILETIME, HANDLE, STILL_ACTIVE};
    use windows::Win32::System::Threading::{
        GetExitCodeProcess, GetProcessTimes, OpenProcess, TerminateProcess,
        CREATE_NEW_PROCESS_GROUP, PROCESS_QUERY_INFORMATION, PROCESS_QUERY_LIMITED_INFORMATION,
        PROCESS_TERMINATE,
    };

    /// Send CTRL_BREAK_EVENT to a process group for graceful cancellation.
    ///
    /// Mirrors Python's `_signal_win_subprocess.py`: detach from current console,
    /// attach to the target's console, send CTRL_BREAK, then re-attach to our own.
    ///
    /// Note: When running as a Windows service (Session 0), console manipulation
    /// doesn't work reliably. In that case we return false so the caller falls
    /// back to terminate (immediate kill).
    fn send_ctrl_break(pid: u32) -> bool {
        use windows::Win32::System::Console::{
            AttachConsole, FreeConsole, GenerateConsoleCtrlEvent, CTRL_BREAK_EVENT,
        };

        // Console APIs don't work from Session 0 (Windows services).
        // Fall back to terminate for reliable cancellation.
        if crate::win32::is_session_zero() {
            log::info!(target: "openjd.sessions", "Running in Session 0, skipping CTRL_BREAK (will fall back to terminate)");
            return false;
        }

        unsafe {
            // Detach from our console
            let _ = FreeConsole();
            // Attach to the target process's console
            if AttachConsole(pid).is_err() {
                // Re-attach to parent if we can't attach to target
                let _ = AttachConsole(u32::MAX); // ATTACH_PARENT_PROCESS
                return false;
            }
            let ok = GenerateConsoleCtrlEvent(CTRL_BREAK_EVENT, pid).is_ok();
            // Detach from target and re-attach to parent
            let _ = FreeConsole();
            let _ = AttachConsole(u32::MAX);
            ok
        }
    }

    /// Check if a process is still alive.
    #[allow(dead_code)]
    fn is_process_alive(pid: u32) -> bool {
        unsafe {
            let handle = OpenProcess(PROCESS_QUERY_INFORMATION, false, pid);
            if let Ok(h) = handle {
                let mut code = 0u32;
                let _ = GetExitCodeProcess(h, &mut code);
                let _ = CloseHandle(h);
                code == STILL_ACTIVE.0 as u32
            } else {
                false
            }
        }
    }

    /// The current system wall clock as FILETIME ticks (100-ns intervals since
    /// 1601-01-01 UTC) — the same epoch `GetProcessTimes` reports process
    /// creation times in, so a value from here is directly comparable to a
    /// process creation time.
    ///
    /// Derived from the standard-library system clock rather than
    /// `GetSystemTimeAsFileTime` so no additional `windows` crate feature (and
    /// thus no Cargo.toml change) is required; both read the same UTC system
    /// clock, so the value and wall-clock base are equivalent.
    fn system_time_as_filetime_ticks() -> u64 {
        // 11_644_473_600 seconds separate the FILETIME epoch (1601) from the
        // Unix epoch (1970); 10_000_000 FILETIME ticks make one second.
        const SECS_1601_TO_1970: u64 = 11_644_473_600;
        let now = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap_or_default();
        now.as_secs()
            .saturating_add(SECS_1601_TO_1970)
            .saturating_mul(10_000_000)
            .saturating_add((now.subsec_nanos() / 100) as u64)
    }

    /// Snapshot all `(pid, parent_pid)` pairs in one toolhelp pass, together
    /// with a wall-clock stamp captured immediately AFTER
    /// `CreateToolhelp32Snapshot` succeeds (before the enumeration walk).
    ///
    /// `TH32CS_SNAPPROCESS` copies the process list at call time;
    /// `Process32First`/`Next` iterate that frozen copy, so nothing created
    /// after the call can appear in the pairs. A stamp taken right after the
    /// call therefore upper-bounds the creation time of every listed process
    /// AND keeps the reuse-acceptance window at its smallest sound value.
    /// `collect_tree_validated` uses it to reject a candidate whose freshly
    /// read creation time is newer than the stamp: the signature of a PID
    /// reused after the snapshot was taken. Taking the stamp after the walk
    /// (as an earlier version did) only widened that window by the walk's
    /// duration for no benefit, since the walk cannot observe any process the
    /// frozen copy did not already contain.
    fn snapshot_process_parents() -> (u64, Vec<(u32, u32)>) {
        use windows::Win32::System::Diagnostics::ToolHelp::{
            CreateToolhelp32Snapshot, Process32FirstW, Process32NextW, PROCESSENTRY32W,
            TH32CS_SNAPPROCESS,
        };
        let mut pairs = Vec::new();
        // Stamp is captured immediately after the snapshot is created (below);
        // this 0 is only the unreachable no-snapshot fallback.
        let mut stamp: u64 = 0;
        unsafe {
            let snap = CreateToolhelp32Snapshot(TH32CS_SNAPPROCESS, 0);
            if let Ok(snap) = snap {
                // Point-in-time: CreateToolhelp32Snapshot froze the process
                // list on the line above, so a stamp taken here upper-bounds
                // every listed process while keeping the reuse-acceptance
                // window at its smallest sound value.
                stamp = system_time_as_filetime_ticks();
                let mut entry = PROCESSENTRY32W {
                    dwSize: std::mem::size_of::<PROCESSENTRY32W>() as u32,
                    ..Default::default()
                };
                if Process32FirstW(snap, &mut entry).is_ok() {
                    loop {
                        pairs.push((entry.th32ProcessID, entry.th32ParentProcessID));
                        if Process32NextW(snap, &mut entry).is_err() {
                            break;
                        }
                    }
                }
                let _ = CloseHandle(snap);
            }
        }
        (stamp, pairs)
    }

    /// Read a process's creation time from an already-open handle.
    ///
    /// The handle must grant at least `PROCESS_QUERY_LIMITED_INFORMATION`.
    /// Packs the creation `FILETIME` into a single `u64` tick count
    /// (`high << 32 | low`). Returns `None` if `GetProcessTimes` fails.
    fn creation_time_from_handle(handle: HANDLE) -> Option<u64> {
        let mut creation = FILETIME::default();
        let mut exit = FILETIME::default();
        let mut kernel = FILETIME::default();
        let mut user = FILETIME::default();
        let ok = unsafe {
            // SAFETY: `handle` is a live process handle supplied by the
            // caller (opened via a successful OpenProcess) granting at least
            // PROCESS_QUERY_LIMITED_INFORMATION. All four out-params are
            // valid, writable FILETIME slots on this stack frame; we only
            // consume `creation`.
            GetProcessTimes(handle, &mut creation, &mut exit, &mut kernel, &mut user).is_ok()
        };
        if !ok {
            return None;
        }
        Some(((creation.dwHighDateTime as u64) << 32) | creation.dwLowDateTime as u64)
    }

    /// Read the creation time of the process currently at `pid`.
    ///
    /// Fail closed: any failure to open the process or read its times
    /// returns `None`, and callers treat `None` as "cannot validate, do not
    /// kill". `PROCESS_QUERY_LIMITED_INFORMATION` can be denied on protected
    /// (PPL) processes, in which case the candidate is skipped: a
    /// fail-closed under-kill by design.
    fn process_creation_time(pid: u32) -> Option<u64> {
        unsafe {
            // SAFETY: OpenProcess is called with a valid access mask and
            // returns a handle we own; on success we read its times and
            // CloseHandle it on every path out of this block. On failure the
            // `?` short-circuits before any handle exists to leak.
            let handle = OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION, false, pid).ok()?;
            let ct = creation_time_from_handle(handle);
            let _ = CloseHandle(handle);
            ct
        }
    }

    /// Terminate `target` only if the process currently at its PID still has
    /// the creation time recorded at collection, closing the collect-to-kill
    /// TOCTOU window: a collected PID could be reused by an unrelated process
    /// before we terminate it. A single handle is opened with both query and
    /// terminate rights, and BOTH the identity read and the `TerminateProcess`
    /// use that SAME handle, so no reuse can slip between the check and the
    /// kill. On mismatch or unreadable identity the kill is skipped and the
    /// reason logged so field misfires are diagnosable. Returns whether a
    /// terminate was issued.
    fn kill_process_checked(target: &ValidatedProcess) -> bool {
        unsafe {
            // SAFETY: OpenProcess returns a handle we own, granting both
            // query and terminate rights; it is CloseHandle'd on every
            // return path below (open-failure returns before a handle
            // exists). The same handle backs both the identity read and the
            // TerminateProcess, so no PID reuse can slip between check and
            // kill.
            let Ok(handle) = OpenProcess(
                PROCESS_QUERY_LIMITED_INFORMATION | PROCESS_TERMINATE,
                false,
                target.pid,
            ) else {
                // Cannot open => cannot validate => fail closed.
                log::info!(target: "openjd.sessions", "Not terminating pid {}: could not read creation time (open failed), PID likely reused or process exited", target.pid);
                return false;
            };
            let current = creation_time_from_handle(handle);
            if !process_identity_matches(target.creation_time, current) {
                let reason = if current.is_some() {
                    "creation time mismatch, PID likely reused"
                } else {
                    "could not read creation time"
                };
                log::info!(target: "openjd.sessions", "Not terminating pid {}: {reason}", target.pid);
                let _ = CloseHandle(handle);
                return false;
            }
            let issued = TerminateProcess(handle, 1).is_ok();
            let _ = CloseHandle(handle);
            issued
        }
    }

    /// Kill a process tree: collect all descendants (validating every
    /// creation-time edge), then kill leaf-to-root. Mirrors Python's
    /// `_windows_process_killer.py`.
    ///
    /// `th32ParentProcessID` is recorded at child creation time, so PID
    /// reuse can make the recorded parent graph cyclic (a dead ancestor's
    /// PID reused by a descendant). The previous recursive implementation
    /// had no cycle guard and overflowed the thread's stack (0xc00000fd)
    /// when it hit such a cycle. Edge validation now lives in
    /// `super::collect_tree_validated`; this history is kept here because
    /// this is the platform entry point that walks that graph.
    fn kill_process_tree(root_pid: u32, expected_root_ct: Option<u64>) {
        let (snapshot_time, parents) = snapshot_process_parents();
        let tree = super::collect_tree_validated(
            root_pid,
            &parents,
            snapshot_time,
            expected_root_ct,
            &mut |pid| process_creation_time(pid),
        );
        if tree.is_empty() {
            // The root's identity could not be established: no pinned identity
            // and an unreadable fresh read, or a pinned identity that no longer
            // matches the fresh read. A raw-PID kill here would risk
            // terminating a reused PID, and it would not even reliably widen
            // coverage: kill_process_checked opens
            // PROCESS_QUERY_LIMITED_INFORMATION | PROCESS_TERMINATE as a single
            // all-or-nothing OpenProcess, so a DACL granting terminate but not
            // query now fails to open where an old TERMINATE-only kill would
            // have succeeded. That narrowing is a deliberate trade: the
            // same-handle identity read is what closes the check-to-kill
            // window, and giving it up reopens the TOCTOU this code exists to
            // eliminate. Fail closed: log and return without killing.
            log::info!(target: "openjd.sessions", "Not terminating tree for pid {root_pid}: root identity could not be established (process exited or protected), skipping kill");
            return;
        }
        // collect_tree_validated returns breadth-first order (every parent
        // before its children), so the reverse kills leaves before ancestors.
        for target in tree.iter().rev() {
            kill_process_checked(target);
        }
    }

    /// Terminate: kill the entire process tree.
    pub fn send_terminate(pid: i32) {
        // Immediate path: the caller still holds the live `Child` handle, so
        // the kernel pins this root PID and it cannot be reused out from under
        // us. There is no separate schedule-time identity to thread, so pass
        // `None` and let collect_tree_validated read the root fresh.
        kill_process_tree(pid as u32, None);
    }

    /// Notify: send CTRL_BREAK_EVENT for graceful shutdown.
    pub fn send_notify(pid: i32) {
        if !send_ctrl_break(pid as u32) {
            log::warn!(target: "openjd.sessions", "Failed to send CTRL_BREAK to pid {pid}, falling back to terminate");
            send_terminate(pid);
        }
    }

    /// Delayed terminate: kill the process tree after a grace period.
    ///
    /// Captures the root's creation time NOW, synchronously, while the caller
    /// still holds the live `Child` handle so the kernel pins this PID and it
    /// cannot be reused out from under us. That schedule-time identity is then
    /// threaded into the kill: `kill_process_tree` re-reads the root after the
    /// delay and fails closed unless it still reports exactly this creation
    /// time, so a PID recycled while the task was sleeping is never killed.
    pub fn spawn_delayed_terminate(pid: i32, delay: Duration) {
        let root_identity = process_creation_time(pid as u32);
        tokio::spawn(async move {
            tokio::time::sleep(delay).await;
            match root_identity {
                Some(ct) => {
                    // Thread the schedule-time identity into the kill. This
                    // replaces the old separate gate (read identity, then call
                    // an unpinned kill): collect_tree_validated now re-reads
                    // the root and returns empty unless it still reports `ct`,
                    // so reuse between the check and the walk cannot slip
                    // through. Each node is revalidated again at kill time
                    // inside kill_process_checked.
                    kill_process_tree(pid as u32, Some(ct));
                }
                None => {
                    // Root creation time could not be queried at schedule
                    // time. This does NOT mean the process had exited: the
                    // caller still held the live `Child` handle here, which
                    // keeps the kernel process object (and its PID) valid even
                    // for an already-terminated process, and a dead-but-held
                    // process still reports its creation time. So a `None` read
                    // means the open/query was DENIED (e.g. a protected
                    // process), not that the process is gone. Either way there
                    // is no pinned identity to validate against, so the safe
                    // action is not to walk the tree at all rather than risk a
                    // fresh unpinned read matching a reused PID.
                    log::warn!(target: "openjd.sessions", "Delayed terminate for pid {pid}: root creation time could not be queried at schedule time (access denied?); no identity to validate against, so the tree will not be walked after the grace period");
                }
            }
        });
    }

    /// Configure the Command for Windows: CREATE_NEW_PROCESS_GROUP + merge stderr into stdout.
    ///
    /// Creates a single OS pipe and sets both stdout and stderr to the write end,
    /// mirroring POSIX `dup2(1, 2)`. Returns the read end as an async reader.
    ///
    /// # Safety
    /// This function is safe on Windows (no pre_exec).
    pub unsafe fn configure_command(
        cmd: &mut Command,
        _use_setsid: bool,
    ) -> Option<Box<dyn tokio::io::AsyncRead + Unpin + Send>> {
        use std::os::windows::io::{FromRawHandle, OwnedHandle};
        use windows::Win32::Security::SECURITY_ATTRIBUTES;
        use windows::Win32::System::Pipes::CreatePipe;

        // CREATE_NEW_PROCESS_GROUP is required for CTRL_BREAK_EVENT to work
        cmd.creation_flags(CREATE_NEW_PROCESS_GROUP.0);

        // Create an anonymous pipe: read_handle for us, write_handle for the child
        let mut read_handle = HANDLE::default();
        let mut write_handle = HANDLE::default();
        let sa = SECURITY_ATTRIBUTES {
            nLength: std::mem::size_of::<SECURITY_ATTRIBUTES>() as u32,
            bInheritHandle: true.into(),
            lpSecurityDescriptor: std::ptr::null_mut(),
        };
        if CreatePipe(&mut read_handle, &mut write_handle, Some(&sa), 0).is_err() {
            // Fall back to separate pipes
            cmd.stdout(std::process::Stdio::piped());
            cmd.stderr(std::process::Stdio::piped());
            return None;
        }

        // Convert write handle to Stdio for the child process.
        // We need two copies: one for stdout, one for stderr.
        let write_owned = OwnedHandle::from_raw_handle(write_handle.0);
        let write_stdio_stdout = std::process::Stdio::from(write_owned);

        // Duplicate the write handle for stderr
        use windows::Win32::Foundation::DuplicateHandle;
        use windows::Win32::System::Threading::GetCurrentProcess;
        let mut write_handle_dup = HANDLE::default();
        let current_process = GetCurrentProcess();
        if DuplicateHandle(
            current_process,
            write_handle,
            current_process,
            &mut write_handle_dup,
            0,
            true, // bInheritHandle
            windows::Win32::Foundation::DUPLICATE_SAME_ACCESS,
        )
        .is_err()
        {
            // Fall back: just use the one handle for stdout, pipe stderr separately
            cmd.stdout(write_stdio_stdout);
            cmd.stderr(std::process::Stdio::piped());
            let read_owned = OwnedHandle::from_raw_handle(read_handle.0);
            let read_std: std::fs::File = std::fs::File::from(read_owned);
            let read_tokio = tokio::fs::File::from_std(read_std);
            return Some(Box::new(read_tokio));
        }
        let write_owned_dup = OwnedHandle::from_raw_handle(write_handle_dup.0);
        let write_stdio_stderr = std::process::Stdio::from(write_owned_dup);

        cmd.stdout(write_stdio_stdout);
        cmd.stderr(write_stdio_stderr);

        // Convert read handle to an async reader
        let read_owned = OwnedHandle::from_raw_handle(read_handle.0);
        let read_std: std::fs::File = std::fs::File::from(read_owned);
        let read_tokio = tokio::fs::File::from_std(read_std);
        Some(Box::new(read_tokio))
    }
}

use platform::*;

/// A process node whose creation time was successfully read and validated
/// against its parent during tree collection.
#[cfg(any(windows, test))]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
struct ValidatedProcess {
    pid: u32,
    creation_time: u64,
}

/// Collect the process tree rooted at `root_pid` in breadth-first order,
/// validating every parent->child edge against process creation times.
///
/// Walks the `parents` snapshot (a list of `(pid, ppid)` pairs) with a
/// visited set instead of recursing. Both details matter:
/// `th32ParentProcessID` is recorded at child creation time, so PID reuse
/// can make the recorded parent graph cyclic (a dead ancestor's PID reused
/// by a descendant). The previous recursive implementation had no cycle
/// guard and overflowed the thread's stack (0xc00000fd) when it hit such a
/// cycle.
///
/// On top of the cycle guard this mirrors psutil's `Process.children()`
/// PID-reuse guard, which the original Python->Rust port dropped: under a
/// forward-moving system clock, a true child is never older than its
/// parent, so an edge (child, ppid) is accepted only when `ppid` is an
/// already-accepted node AND the child's creation time is `>=` that
/// parent's. A stale edge (the parent PID was reused after the real parent
/// exited) has a child older than the recorded parent and always fails the
/// check. We fail closed on stale edges (a child older than its recorded
/// parent, the mark of a reused parent PID). A child whose creation time is
/// unreadable is NOT skipped: it is recorded as a traversable node carrying
/// the parent's creation time as its bound (see the later paragraph), so a
/// live subtree under a dead intermediate is still reached. If the ROOT's
/// creation time is unreadable we return an empty `Vec` ONLY when no
/// schedule-time identity is pinned; with `Some(exp)` the walk is seeded from
/// that pinned identity and proceeds.
///
/// Because `FILETIME` creation times are wall-clock rather than monotonic,
/// this invariant is only exact under a forward-moving clock. An NTP
/// step-back or manual clock change between a parent's and its child's
/// creation can make a TRUE child's recorded creation time earlier than its
/// parent's, failing the `>=` edge check and pruning that child's whole
/// subtree. The failure mode is a leaked (skipped) descendant, never
/// terminating an unrelated process, so this stays on the safe under-kill
/// side.
///
/// `snapshot_time` closes a second PID-reuse direction: the one where a
/// genuine edge (child, parent) is captured in the snapshot, the child then
/// exits, and its PID is reused by an unrelated process U before this function
/// reads the child's creation time. U's creation time is newer than the
/// parent's, so the `>=` edge check passes and U would be recorded as a
/// validated node. Because every process the snapshot lists was created before
/// the enumeration walk finished (the stamp is captured after the walk), we
/// reject any DESCENDANT candidate whose freshly read creation time is greater
/// than `snapshot_time`. The stamp and process creation times share the same
/// wall-clock base (FILETIME ticks since 1601 UTC), so the comparison is exact.
///
/// The root is exempt from the `snapshot_time` bound in every arm. The root's
/// identity is never established by fossil-edge attribution (the reuse the
/// stamp guards): it is either pinned by a live `Child` handle (the immediate
/// `send_terminate` path, `expected_root_ct = None`), matched against a
/// schedule-time identity (`Some(exp)` with `read == exp`), or seeded from that
/// pinned identity when the root has already exited (`Some(exp)` with `None`).
/// A stamp comparison on the root could therefore never catch a reused PID; it
/// could only false-positive under a backward wall-clock adjustment (an NTP
/// step-back or a VM resume, where the root's FILETIME creation was recorded
/// before the step and the stamp read after it), return empty, and turn a
/// legitimate cancel into a complete no-op — nothing killed, the session left
/// waiting on a live tree. Accepted degradation: under such a backward step the
/// root is still killed, but descendants whose recorded creation times exceed
/// the post-step stamp may be skipped. That is an under-kill (a leaked
/// descendant), which stays on the safe side.
///
/// `expected_root_ct` pins the root's identity to a value established before
/// this call (a schedule-time read from the delayed-terminate path). When
/// `Some(exp)`:
/// - a fresh root read of `Some(rc)` must equal `exp`, otherwise this returns
///   empty (fail closed) — closing the reuse window between an earlier identity
///   read and this walk;
/// - a fresh root read of `None` (the root exited during the grace period, its
///   PID currently unassigned) SEEDS the walk with `ValidatedProcess { pid:
///   root_pid, creation_time: exp }` and proceeds. This is the common
///   delayed-path case, whose whole purpose is reaping orphaned grandchildren:
///   `kill_process_checked` no-ops on the dead root, but its orphaned
///   descendants are still enumerated and validated against `>= exp` and
///   `<= snapshot_time`.
///
/// The recorded root node then carries `exp`, the pinned identity, not any
/// re-read value. When `expected_root_ct` is `None`, an unreadable fresh root
/// read returns empty (fail closed) and a readable one is accepted as-is (the
/// live `Child` handle already pins the PID; see the root-exemption note
/// above).
///
/// A child whose creation time is unreadable under an already-validated parent
/// is kept TRAVERSABLE rather than pruned: it is recorded carrying the parent's
/// creation time (a sound lower bound for ITS children) so that a live subtree
/// under a dead intermediate (cmd -> sh -> renderer, sh exits first) is still
/// reached. `kill_process_checked` cannot open the dead PID, so nothing is
/// terminated for the node itself; only its readable, validated descendants
/// are. Stale (`< parent`) and post-snapshot (`> snapshot_time`) rejections
/// still apply to READABLE children exactly as before.
///
/// Residual: seeding a dead root and traversing dead intermediates reopens a
/// bounded window. A freed PID's fossil edges (recorded `th32ParentProcessID`
/// values) can include children of an INTERIM holder of that PID whose creation
/// times happen to land in `[exp, snapshot_time]`; those pass validation and
/// may be terminated. The window is bounded and no worse than the
/// pre-validation walk, and is structurally eliminated only by the Job Object
/// rework (issue #347).
///
/// Separate residual (distinct from the dead-root/interim-holder case above):
/// an unreadable-but-ALIVE fossil child. If a recorded `th32ParentProcessID`
/// was freed and reassigned to OUR root, an access-denied process carrying
/// that stale parent value becomes a child edge of the root. Because it is
/// unreadable it bypasses the `>=` staleness check by construction (that check
/// only applies to READABLE children) and is recorded as a traversable node
/// carrying the parent's bound, so its own live children can fall inside
/// `[parent_ct, snapshot_time]` and be terminated. Distinguishing genuinely
/// dead PIDs (`OpenProcess` -> `ERROR_INVALID_PARAMETER`) from merely denied
/// ones (`ERROR_ACCESS_DENIED`) would close this, but that changes the
/// creation-time callback contract, so it is deferred with the Job Object
/// rework (issue #347).
#[cfg(any(windows, test))]
fn collect_tree_validated(
    root_pid: u32,
    parents: &[(u32, u32)],
    snapshot_time: u64,
    expected_root_ct: Option<u64>,
    creation_time: &mut dyn FnMut(u32) -> Option<u64>,
) -> Vec<ValidatedProcess> {
    use std::collections::HashSet;

    // Resolve the root's identity. The four cases below implement the pinned
    // vs. unpinned, readable vs. exited matrix documented above.
    let root_ct = match (expected_root_ct, creation_time(root_pid)) {
        // Pinned identity, root still readable: the fresh read must match the
        // pin exactly (else reuse/exit). No snapshot_time bound — the identity
        // match already proves this is the pinned process, so a stamp
        // comparison here could only false-positive under a backward clock
        // step-back and turn a legitimate cancel into a no-op.
        //
        // On mismatch we deliberately ABORT (return empty) rather than seed the
        // walk. If `rc != exp` the PID has been REASSIGNED: the new occupant is
        // alive, and its children are guaranteed-live unrelated processes whose
        // creation times necessarily land in `[exp, snapshot_time]`, so they
        // would pass the `>=`/`<=` edge checks. Seeding a root here would
        // therefore deterministically kill those unrelated processes. Aborting
        // instead only leaks our own orphans (under-kill), which is the safe
        // direction.
        (Some(exp), Some(rc)) => {
            if rc != exp {
                return Vec::new();
            }
            exp
        }
        // Pinned identity, root exited during the grace period: seed with the
        // pinned identity so orphaned descendants are still reaped.
        (Some(exp), None) => exp,
        // No pinned identity and no fresh read: nothing to validate against.
        (None, None) => return Vec::new(),
        // No pinned identity, root readable: the only caller passing None is the
        // immediate send_terminate path, where the live Child handle pins the
        // PID, so rc cannot belong to a reused PID. No snapshot_time bound — a
        // stamp comparison here can only false-positive under a backward clock
        // adjustment and would turn a legitimate cancel into a no-op.
        (None, Some(rc)) => rc,
    };

    let mut visited: HashSet<u32> = HashSet::from([root_pid]);
    let mut result: Vec<ValidatedProcess> = vec![ValidatedProcess {
        pid: root_pid,
        creation_time: root_ct,
    }];

    let mut i = 0;
    while i < result.len() {
        let parent = result[i];
        i += 1;
        for &(child, ppid) in parents {
            // Only descend from the node we are currently processing, and
            // visit each child once (this is the cycle/duplicate guard).
            if ppid != parent.pid || !visited.insert(child) {
                continue;
            }
            match creation_time(child) {
                Some(child_ct) => {
                    // A true child is never older than its parent, and — like
                    // every genuine snapshot member — was created before the
                    // snapshot. A creation time newer than the snapshot marks a
                    // PID reused after the snapshot (fail closed); an
                    // older-than-parent one marks a stale reused-parent edge.
                    if child_ct >= parent.creation_time && child_ct <= snapshot_time {
                        result.push(ValidatedProcess {
                            pid: child,
                            creation_time: child_ct,
                        });
                    }
                }
                None => {
                    // Unreadable child under a validated parent: the node
                    // itself cannot be opened (kill_process_checked no-ops on
                    // it), but it may still have live descendants. Keep it
                    // traversable carrying the parent's creation time — a sound
                    // lower bound for ITS children — instead of pruning the
                    // whole subtree. See the interim-holder residual note above.
                    result.push(ValidatedProcess {
                        pid: child,
                        creation_time: parent.creation_time,
                    });
                }
            }
        }
    }
    result
}

/// Kill-time revalidation predicate guarding the TOCTOU window between tree
/// collection and `TerminateProcess`: a PID collected earlier could be
/// reused by an unrelated process before we terminate it. Returns `true`
/// only when the process currently at that PID reports exactly the creation
/// time recorded at collection.
#[cfg(any(windows, test))]
fn process_identity_matches(expected_creation_time: u64, current: Option<u64>) -> bool {
    current == Some(expected_creation_time)
}

/// Write cancel_info.json to the working directory as required by the OpenJD spec
/// for NotifyThenTerminate cancelation.
fn write_cancel_info(working_dir: &Path, terminate_delay: Duration) {
    let notify_end = std::time::SystemTime::now() + terminate_delay;
    let secs = notify_end
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap_or_default()
        .as_secs();
    // Format as ISO 8601 UTC
    let s = secs % 60;
    let m = (secs / 60) % 60;
    let h = (secs / 3600) % 24;
    let total_days = secs / 86400;
    // Simple date calculation from days since epoch
    let (y, mo, d) = days_to_ymd(total_days);
    let timestamp = format!("{y:04}-{mo:02}-{d:02}T{h:02}:{m:02}:{s:02}Z");
    let info = serde_json::json!({ "NotifyEnd": timestamp });
    let path = working_dir.join("cancel_info.json");
    let _ = std::fs::write(&path, serde_json::to_string(&info).unwrap_or_default());
}

/// Convert days since Unix epoch to (year, month, day).
fn days_to_ymd(total_days: u64) -> (u64, u64, u64) {
    // Algorithm from http://howardhinnant.github.io/date_algorithms.html
    let z = total_days + 719468;
    let era = z / 146097;
    let doe = z - era * 146097;
    let yoe = (doe - doe / 1460 + doe / 36524 - doe / 146096) / 365;
    let y = yoe + era * 400;
    let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
    let mp = (5 * doy + 2) / 153;
    let d = doy - (153 * mp + 2) / 5 + 1;
    let m = if mp < 10 { mp + 3 } else { mp - 9 };
    let y = if m <= 2 { y + 1 } else { y };
    (y, m, d)
}

/// Format a command argument list for logging, applying redaction for any
/// `openjd_redacted_env:` tokens that may appear in the arguments.
pub(crate) fn format_command_for_log(args: &[String]) -> String {
    let joined =
        shlex::try_join(args.iter().map(|s| s.as_str())).unwrap_or_else(|_| args.join(" "));
    crate::action_filter::redact_openjd_redacted_env_requests(&joined)
}

/// Run a subprocess asynchronously with real-time stdout streaming through an ActionFilter.
///
/// Spawns the process with merged environment variables, streams stdout line-by-line
/// through the ActionFilter, supports cancellation and timeout, uses process groups
/// (setsid) for proper signal delivery, and handles stdout grace time for detached
/// grandchild processes.
///
/// Parsed `ActionMessage` values are sent through `message_tx` in real-time as
/// stdout lines are processed.
pub async fn run_subprocess(
    config: SubprocessConfig,
    filter: &mut ActionFilter,
    session_id: &str,
    message_tx: mpsc::UnboundedSender<ActionMessage>,
    cancel_token: CancellationToken,
) -> Result<SubprocessResult, SessionError> {
    let args = &config.args;
    if args.is_empty() {
        return Err(SessionError::Runtime("No command specified".into()));
    }

    // Cross-user execution must go through the helper binary.
    if config.user.as_deref().is_some_and(|u| !u.is_process_user()) {
        return Err(SessionError::Runtime(
            "Cross-user subprocess execution requires the helper binary. \
             Use run_via_helper instead of run_subprocess for cross-user actions."
                .into(),
        ));
    }

    // Windows: resolve the command to an absolute path with canonical
    // search semantics (PATHEXT-aware, working_dir first, action PATH) so
    // the spawn below cannot fall back to CreateProcessW's legacy search.
    // See win32_locate.rs. Mirrors Python's locate_windows_executable call
    // in _runner_base.py.
    #[cfg(windows)]
    let args = &{
        let wd = config
            .working_dir
            .clone()
            .unwrap_or_else(|| std::path::PathBuf::from("."));
        crate::win32_locate::locate_windows_executable(args, Some(&config.env_vars), &wd).map_err(
            |msg| {
                session_log!(
                    info,
                    session_id,
                    LogContent::EXCEPTION_INFO | LogContent::PROCESS_CONTROL,
                    "{}",
                    msg
                );
                SessionError::SubprocessStart {
                    command: args[0].clone(),
                    source: std::io::Error::new(std::io::ErrorKind::NotFound, msg),
                }
            },
        )?
    };

    // Build merged environment
    let mut merged: HashMap<String, String> = std::env::vars().collect();
    for (k, v) in &config.env_vars {
        match v {
            Some(val) => {
                merged.insert(k.clone(), val.clone());
            }
            None => {
                merged.remove(k);
            }
        }
    }

    // Log the command line (redacting any openjd_redacted_env tokens)
    session_log!(
        info,
        session_id,
        LogContent::FILE_PATH | LogContent::PROCESS_CONTROL,
        "Running command {}",
        format_command_for_log(args)
    );

    // Spawn the process via tokio::process::Command (same-user only;
    // cross-user was rejected above).
    #[cfg(windows)]
    let win32_process_handle: Option<windows::Win32::Foundation::HANDLE> = None;

    #[allow(unused_mut)]
    let (mut child, pid, stdout_for_reading): (
        Option<tokio::process::Child>,
        i32,
        Option<Box<dyn tokio::io::AsyncRead + Unpin + Send>>,
    ) = {
        let mut cmd = Command::new(&args[0]);
        cmd.args(&args[1..]);
        cmd.env_clear();
        for (k, v) in &merged {
            cmd.env(k, v);
        }
        if let Some(dir) = &config.working_dir {
            cmd.current_dir(dir);
        }
        let merged_reader = unsafe { configure_command(&mut cmd, true) };
        if merged_reader.is_none() {
            cmd.stdout(std::process::Stdio::piped());
        }
        let mut c = cmd.spawn().map_err(|e| {
            session_log!(
                info,
                session_id,
                LogContent::EXCEPTION_INFO | LogContent::PROCESS_CONTROL,
                "Process failed to start: '{}': {}",
                args[0],
                e
            );
            SessionError::SubprocessStart {
                command: args[0].clone(),
                source: e,
            }
        })?;
        let p = c.id().unwrap_or(0) as i32;
        let stdout = merged_reader.or_else(|| {
            c.stdout
                .take()
                .map(|s| Box::new(s) as Box<dyn tokio::io::AsyncRead + Unpin + Send>)
        });
        (Some(c), p, stdout)
    };

    session_log!(
        info,
        session_id,
        LogContent::PROCESS_CONTROL,
        "Command started as pid: {}",
        pid
    );
    session_log!(
        info,
        session_id,
        LogContent::BANNER | LogContent::COMMAND_OUTPUT,
        "Output:"
    );

    // Read merged stdout+stderr from the child
    let mut cancel_requested = false;
    let mut timed_out = false;
    // True once we've issued `send_terminate` on the process tree from
    // inside the stdout read loop (timeout path, urgent cancel, or
    // `CancelMethod::Terminate`). `send_terminate` is the crate's
    // platform-agnostic "kill now" — SIGKILL on Unix, `TerminateProcess`
    // via `kill_process_tree` on Windows.
    //
    // Stays false for `NotifyThenTerminate` cancel — there the terminate
    // is only scheduled (via `spawn_delayed_terminate`), so the process
    // may still be winding down gracefully in response to the notify
    // signal when the loop exits and needs the longer `STDOUT_GRACE_TIME`
    // on the final `c.wait()`.
    let mut terminate_sent = false;
    let mut stdout_collected = String::new();
    let mut saw_fail = false;

    if let Some(stdout) = stdout_for_reading {
        let mut reader = BufReader::new(stdout);
        let mut line_buf = Vec::new();

        // Create timeout future once, pin it for reuse across loop iterations
        let timeout_fut = async {
            match config.timeout {
                Some(d) => tokio::time::sleep(d).await,
                None => std::future::pending().await,
            }
        };
        tokio::pin!(timeout_fut);

        // Grace period for stdout to drain after process termination.
        // On Windows, killed processes (especially MSYS2 sh.exe) may leave
        // inherited pipe handles open in orphaned child processes, preventing
        // EOF. This deadline ensures we don't hang forever.
        let drain_deadline = tokio::time::sleep(Duration::MAX);
        tokio::pin!(drain_deadline);

        loop {
            tokio::select! {
                biased;

                _ = &mut drain_deadline, if cancel_requested => {
                    session_log!(info, session_id, LogContent::PROCESS_CONTROL,
                        "Stdout drain grace period expired, stopping read loop");
                    break;
                }

                _ = cancel_token.cancelled(), if !cancel_requested => {
                    cancel_requested = true;
                    drain_deadline.as_mut().reset(tokio::time::Instant::now() + STDOUT_DRAIN_AFTER_KILL);
                    let time_limit = config.cancel_request_rx.as_ref()
                        .and_then(|rx| *rx.borrow());

                    match (&config.cancel_method, time_limit) {
                        (_, Some(limit)) if limit.is_zero() => {
                            session_log!(info, session_id, LogContent::PROCESS_CONTROL, "Urgent cancel (time_limit=0), sending SIGKILL to process group {}", pid);
                            send_terminate(pid);
                            terminate_sent = true;
                        }
                        (CancelMethod::Terminate, _) => {
                            session_log!(info, session_id, LogContent::PROCESS_CONTROL, "Sending SIGKILL to process group {}", pid);
                            send_terminate(pid);
                            terminate_sent = true;
                        }
                        (CancelMethod::NotifyThenTerminate { terminate_delay }, _) => {
                            let delay = match time_limit {
                                Some(limit) => limit.min(*terminate_delay),
                                None => *terminate_delay,
                            };
                            if let Some(dir) = &config.working_dir {
                                write_cancel_info(dir, delay);
                            }
                            session_log!(info, session_id, LogContent::PROCESS_CONTROL, "Sending SIGTERM to process group {} (grace period: {:?})", pid, delay);
                            send_notify(pid);
                            spawn_delayed_terminate(pid, delay);
                            // Deliberately do NOT set terminate_sent here —
                            // the terminate is only scheduled (via
                            // `spawn_delayed_terminate`), not yet delivered.
                            // The process may still be exiting gracefully in
                            // response to the notify signal, and the final
                            // `c.wait()` should get the longer
                            // `STDOUT_GRACE_TIME`.
                        }
                    }
                }

                _ = &mut timeout_fut, if !cancel_requested && !timed_out => {
                    timed_out = true;
                    cancel_requested = true;
                    drain_deadline.as_mut().reset(tokio::time::Instant::now() + STDOUT_DRAIN_AFTER_KILL);
                    session_log!(info, session_id, LogContent::PROCESS_CONTROL, "Action timed out, sending SIGKILL to process group");
                    send_terminate(pid);
                    terminate_sent = true;
                }

                n = reader.read_until(b'\n', &mut line_buf) => {
                    match n {
                        Ok(0) => break, // EOF
                        Ok(_) => {
                            // Strip trailing newline (and \r on Windows)
                            if line_buf.last() == Some(&b'\n') {
                                line_buf.pop();
                            }
                            if line_buf.last() == Some(&b'\r') {
                                line_buf.pop();
                            }
                            let line = decode_backslashreplace(&line_buf);
                            let line = truncate_line(&line).to_string();
                            line_buf.clear();
                            let (display, pass_through) = process_line(&line, filter, session_id, &message_tx, &mut saw_fail);
                            if pass_through && filter.min_log_level() <= 20 {
                                session_log!(info, session_id, LogContent::COMMAND_OUTPUT, "{}", display);
                            }
                            if config.debug_collect_stdout {
                                stdout_collected.push_str(&display);
                                stdout_collected.push('\n');
                            }
                        }
                        Err(_) => break,
                    }
                }
            }
        }
    }

    // Wait for process to exit
    let exit_status = if let Some(ref mut c) = child {
        // If we already issued `send_terminate` from inside the read loop
        // (timeout, urgent cancel, or Terminate cancel), the child should
        // be dead and `c.wait()` just needs to reap it — a short bound
        // is plenty. Otherwise give the full 5s to accommodate graceful
        // shutdown (natural EOF or `NotifyThenTerminate`).
        let grace = if terminate_sent {
            STDOUT_GRACE_TIME_POST_TERMINATE
        } else {
            STDOUT_GRACE_TIME
        };
        match tokio::time::timeout(grace, c.wait()).await {
            Ok(Ok(s)) => Some(s),
            Ok(Err(_)) => {
                send_terminate(pid);
                None
            }
            Err(_) => {
                send_terminate(pid);
                c.wait().await.ok()
            }
        }
    } else {
        // Windows cross-user: wait on the raw process handle
        #[cfg(windows)]
        {
            win32_process_handle.map(|h| {
                use std::os::windows::process::ExitStatusExt;
                use windows::Win32::System::Threading::{GetExitCodeProcess, WaitForSingleObject};
                unsafe {
                    let _ = WaitForSingleObject(h, 60000);
                    let mut code = 0u32;
                    let _ = GetExitCodeProcess(h, &mut code);
                    let _ = windows::Win32::Foundation::CloseHandle(h);
                    std::process::ExitStatus::from_raw(code)
                }
            })
        }
        #[cfg(not(windows))]
        {
            None
        }
    };

    let exit_code = exit_status.and_then(|s| s.code());
    session_log!(
        info,
        session_id,
        LogContent::PROCESS_CONTROL,
        "Process exit code: {}",
        exit_code.map_or("N/A".to_string(), |c| c.to_string())
    );

    let state = if timed_out {
        ActionState::Timeout
    } else if cancel_requested || cancel_token.is_cancelled() {
        ActionState::Canceled
    } else if saw_fail {
        ActionState::Failed
    } else if exit_status.is_some_and(|s| s.success()) {
        ActionState::Success
    } else {
        ActionState::Failed
    };

    Ok(SubprocessResult {
        state,
        exit_code,
        stdout: stdout_collected,
    })
}

pub(crate) fn process_line(
    line: &str,
    filter: &mut ActionFilter,
    session_id: &str,
    message_tx: &mpsc::UnboundedSender<ActionMessage>,
    saw_fail: &mut bool,
) -> (String, bool) {
    let (callbacks, pass_through, display) = filter.filter_message(line, session_id);
    for cb in callbacks {
        let cancel = cb.cancel;
        let msg = match cb.kind {
            ActionMessageKind::Progress => {
                if let ActionMessageValue::Float(v) = cb.value {
                    Some(ActionMessage::Progress(v))
                } else {
                    None
                }
            }
            ActionMessageKind::Status => {
                if let ActionMessageValue::String(s) = cb.value {
                    Some(ActionMessage::Status(s))
                } else {
                    None
                }
            }
            ActionMessageKind::Fail => {
                if let ActionMessageValue::String(s) = cb.value {
                    *saw_fail = true;
                    Some(ActionMessage::Fail(s))
                } else {
                    None
                }
            }
            ActionMessageKind::Env => {
                if let ActionMessageValue::EnvVar { name, value } = cb.value {
                    Some(ActionMessage::SetEnv { name, value })
                } else {
                    None
                }
            }
            ActionMessageKind::UnsetEnv => {
                if let ActionMessageValue::String(name) = cb.value {
                    Some(ActionMessage::UnsetEnv { name })
                } else {
                    None
                }
            }
            ActionMessageKind::RedactedEnv => {
                if let ActionMessageValue::EnvVar { name, value } = cb.value {
                    Some(ActionMessage::RedactedEnv { name, value })
                } else {
                    None
                }
            }
            _ => None,
        };
        if let Some(msg) = msg {
            let _ = message_tx.send(msg);
        }
        if cancel {
            let fail_msg = "Action canceled due to malformed command".to_string();
            let _ = message_tx.send(ActionMessage::CancelMarkFailed {
                fail_message: fail_msg,
            });
        }
    }
    (display, pass_through)
}

#[cfg(test)]
mod tests {
    #[allow(unused_imports)]
    use super::*;

    #[cfg(unix)]
    #[tokio::test]
    async fn test_cancel_ntt_with_zero_time_limit_is_immediate() {
        use tokio_util::sync::CancellationToken;

        let token = CancellationToken::new();
        let (_cancel_tx, cancel_rx) = tokio::sync::watch::channel(None);
        let (msg_tx, _msg_rx) = tokio::sync::mpsc::unbounded_channel();

        let config = SubprocessConfig {
            args: vec!["sleep".into(), "30".into()],
            env_vars: HashMap::new(),
            working_dir: None,
            timeout: None,
            user: None,
            cancel_method: CancelMethod::NotifyThenTerminate {
                terminate_delay: Duration::from_secs(60),
            },
            cancel_request_rx: Some(cancel_rx),
            debug_collect_stdout: false,
        };

        let t = token.clone();
        tokio::spawn(async move {
            tokio::time::sleep(Duration::from_millis(200)).await;
            let _ = _cancel_tx.send(Some(Duration::ZERO));
            t.cancel();
        });

        let mut filter = crate::action_filter::ActionFilter::new("test", true, false);
        let start = std::time::Instant::now();
        let result = run_subprocess(config, &mut filter, "test", msg_tx, token)
            .await
            .unwrap();
        let elapsed = start.elapsed();

        assert_eq!(result.state, ActionState::Canceled);
        assert!(
            elapsed < Duration::from_secs(5),
            "took {:?}, expected < 5s",
            elapsed
        );
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn test_cancel_ntt_without_time_limit_uses_default() {
        use tokio_util::sync::CancellationToken;

        let token = CancellationToken::new();
        let (_cancel_tx, cancel_rx) = tokio::sync::watch::channel(None);
        let (msg_tx, _msg_rx) = tokio::sync::mpsc::unbounded_channel();

        // Use python3 to reliably ignore SIGTERM in a single process. Using
        // `sh -c 'trap "" TERM; sleep 30'` is fragile: some sh implementations
        // install a userspace handler for the trap instead of SIG_IGN, so the
        // SIG_IGN inheritance rule does not protect the child `sleep`, and
        // killpg(SIGTERM) kills `sleep` — terminating the script within ms
        // rather than waiting for the 1s SIGKILL.
        //
        // The script writes a sentinel file after installing the handler so
        // the test can wait for readiness before canceling — avoiding a race
        // where SIGTERM arrives before SIG_IGN is installed.
        let ready_dir = tempfile::tempdir().unwrap();
        let ready_path = ready_dir.path().join("ready");
        let py_script = format!(
            "import signal, time, pathlib; signal.signal(signal.SIGTERM, signal.SIG_IGN); pathlib.Path('{}').write_text('ok'); time.sleep(30)",
            ready_path.display()
        );
        let config = SubprocessConfig {
            args: vec!["python3".into(), "-c".into(), py_script],
            env_vars: HashMap::new(),
            working_dir: None,
            timeout: None,
            user: None,
            cancel_method: CancelMethod::NotifyThenTerminate {
                terminate_delay: Duration::from_secs(1),
            },
            cancel_request_rx: Some(cancel_rx),
            debug_collect_stdout: false,
        };

        let t = token.clone();
        let ready_path_clone = ready_path.clone();
        tokio::spawn(async move {
            // Wait for the python process to install its SIGTERM handler
            for _ in 0..100 {
                if ready_path_clone.exists() {
                    break;
                }
                tokio::time::sleep(Duration::from_millis(50)).await;
            }
            t.cancel();
        });

        let mut filter = crate::action_filter::ActionFilter::new("test", true, false);
        let start = std::time::Instant::now();
        let result = run_subprocess(config, &mut filter, "test", msg_tx, token)
            .await
            .unwrap();
        let elapsed = start.elapsed();

        assert_eq!(result.state, ActionState::Canceled);
        // After cancel, the 1s terminate_delay must elapse before SIGKILL
        assert!(
            elapsed >= Duration::from_millis(800),
            "took {:?}, expected >= 800ms",
            elapsed
        );
        assert!(
            elapsed < Duration::from_secs(10),
            "took {:?}, expected < 10s",
            elapsed
        );
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn test_cancel_terminate_ignores_time_limit() {
        use tokio_util::sync::CancellationToken;

        let token = CancellationToken::new();
        let (_cancel_tx, cancel_rx) = tokio::sync::watch::channel(None);
        let (msg_tx, _msg_rx) = tokio::sync::mpsc::unbounded_channel();

        let config = SubprocessConfig {
            args: vec!["sleep".into(), "30".into()],
            env_vars: HashMap::new(),
            working_dir: None,
            timeout: None,
            user: None,
            cancel_method: CancelMethod::Terminate,
            cancel_request_rx: Some(cancel_rx),
            debug_collect_stdout: false,
        };

        let t = token.clone();
        tokio::spawn(async move {
            tokio::time::sleep(Duration::from_millis(200)).await;
            let _ = _cancel_tx.send(Some(Duration::from_secs(10)));
            t.cancel();
        });

        let mut filter = crate::action_filter::ActionFilter::new("test", true, false);
        let start = std::time::Instant::now();
        let result = run_subprocess(config, &mut filter, "test", msg_tx, token)
            .await
            .unwrap();
        let elapsed = start.elapsed();

        assert_eq!(result.state, ActionState::Canceled);
        assert!(
            elapsed < Duration::from_secs(2),
            "took {:?}, expected < 2s",
            elapsed
        );
    }

    #[cfg(windows)]
    #[tokio::test]
    async fn test_cancel_terminate_on_windows() {
        use tokio_util::sync::CancellationToken;

        let token = CancellationToken::new();
        let (_cancel_tx, cancel_rx) = tokio::sync::watch::channel(None);
        let (msg_tx, _msg_rx) = tokio::sync::mpsc::unbounded_channel();

        // Use powershell sleep which is a real process (not a shell builtin)
        let config = SubprocessConfig {
            args: vec![
                "powershell".into(),
                "-Command".into(),
                "Start-Sleep 30".into(),
            ],
            env_vars: HashMap::new(),
            working_dir: None,
            timeout: None,
            user: None,
            cancel_method: CancelMethod::Terminate,
            cancel_request_rx: Some(cancel_rx),
            debug_collect_stdout: false,
        };

        let t = token.clone();
        tokio::spawn(async move {
            tokio::time::sleep(Duration::from_millis(500)).await;
            t.cancel();
        });

        let mut filter = crate::action_filter::ActionFilter::new("test", true, false);
        let start = std::time::Instant::now();
        let result = run_subprocess(config, &mut filter, "test", msg_tx, token)
            .await
            .unwrap();
        let elapsed = start.elapsed();

        assert_eq!(result.state, ActionState::Canceled);
        assert!(
            elapsed < Duration::from_secs(5),
            "Cancel took {:?}, expected < 5s — process was not killed promptly",
            elapsed
        );
    }

    #[test]
    fn test_format_command_for_log_simple() {
        let args = vec!["echo".to_string(), "hello".to_string(), "world".to_string()];
        let result = format_command_for_log(&args);
        assert_eq!(result, "echo hello world");
    }

    #[test]
    fn test_format_command_for_log_with_spaces() {
        let args = vec!["echo".to_string(), "hello world".to_string()];
        let result = format_command_for_log(&args);
        // Should be shell-quoted
        assert!(result.contains("hello world"), "got: {result}");
    }

    #[test]
    fn test_format_command_for_log_redacts_secret() {
        let args = vec![
            "python".to_string(),
            "-c".to_string(),
            "print('openjd_redacted_env: PASSWORD=secret123')".to_string(),
        ];
        let result = format_command_for_log(&args);
        assert!(!result.contains("secret123"), "secret leaked in: {result}");
        assert!(
            result.contains("openjd_redacted_env:"),
            "token missing in: {result}"
        );
        assert!(
            result.contains("********"),
            "redaction missing in: {result}"
        );
    }

    #[test]
    fn test_format_command_for_log_no_redaction_needed() {
        let args = vec![
            "python".to_string(),
            "-c".to_string(),
            "print('hello')".to_string(),
        ];
        let result = format_command_for_log(&args);
        assert!(
            result.contains("print('hello')") || result.contains("print"),
            "got: {result}"
        );
        assert!(!result.contains("********"));
    }

    // ── Tier 1: Pure function tests ──────────────────────────────────

    #[test]
    fn test_days_to_ymd_epoch() {
        assert_eq!(days_to_ymd(0), (1970, 1, 1));
    }

    #[test]
    fn test_days_to_ymd_known_date() {
        // 2024-02-29 is a leap day. Days since epoch = 19782
        assert_eq!(days_to_ymd(19782), (2024, 2, 29));
    }

    #[test]
    fn test_days_to_ymd_end_of_year() {
        // 2023-12-31 = day 19722
        assert_eq!(days_to_ymd(19722), (2023, 12, 31));
    }

    #[test]
    fn test_days_to_ymd_y2k() {
        // 2000-01-01 = day 10957
        assert_eq!(days_to_ymd(10957), (2000, 1, 1));
    }

    #[test]
    fn test_write_cancel_info_creates_file() {
        let dir = tempfile::tempdir().unwrap();
        write_cancel_info(dir.path(), Duration::from_secs(30));
        let path = dir.path().join("cancel_info.json");
        assert!(path.exists());
        let content: serde_json::Value =
            serde_json::from_str(&std::fs::read_to_string(&path).unwrap()).unwrap();
        let ts = content["NotifyEnd"].as_str().unwrap();
        assert!(ts.ends_with('Z'), "Expected UTC timestamp, got: {ts}");
        assert!(ts.contains('T'), "Expected ISO 8601, got: {ts}");
    }

    #[test]
    fn test_process_line_plain_text() {
        let (tx, _rx) = mpsc::unbounded_channel();
        let mut filter = ActionFilter::new("test", true, false);
        let mut saw_fail = false;
        let (display, pass_through) =
            process_line("hello world", &mut filter, "test", &tx, &mut saw_fail);
        assert!(pass_through);
        assert_eq!(display, "hello world");
        assert!(!saw_fail);
    }

    #[test]
    fn test_process_line_progress() {
        let (tx, mut rx) = mpsc::unbounded_channel();
        let mut filter = ActionFilter::new("test", true, false);
        let mut saw_fail = false;
        let (_display, _pass_through) = process_line(
            "openjd_progress: 0.5",
            &mut filter,
            "test",
            &tx,
            &mut saw_fail,
        );
        assert!(!saw_fail);
        match rx.try_recv().unwrap() {
            ActionMessage::Progress(v) => assert!((v - 0.5).abs() < f64::EPSILON),
            other => panic!("Expected Progress, got: {other:?}"),
        }
    }

    #[test]
    fn test_process_line_status() {
        let (tx, mut rx) = mpsc::unbounded_channel();
        let mut filter = ActionFilter::new("test", true, false);
        let mut saw_fail = false;
        process_line(
            "openjd_status: rendering frame 42",
            &mut filter,
            "test",
            &tx,
            &mut saw_fail,
        );
        assert!(!saw_fail);
        match rx.try_recv().unwrap() {
            ActionMessage::Status(s) => assert_eq!(s, "rendering frame 42"),
            other => panic!("Expected Status, got: {other:?}"),
        }
    }

    #[test]
    fn test_process_line_fail() {
        let (tx, mut rx) = mpsc::unbounded_channel();
        let mut filter = ActionFilter::new("test", true, false);
        let mut saw_fail = false;
        process_line(
            "openjd_fail: out of memory",
            &mut filter,
            "test",
            &tx,
            &mut saw_fail,
        );
        assert!(saw_fail);
        match rx.try_recv().unwrap() {
            ActionMessage::Fail(s) => assert_eq!(s, "out of memory"),
            other => panic!("Expected Fail, got: {other:?}"),
        }
    }

    #[test]
    fn test_process_line_env() {
        let (tx, mut rx) = mpsc::unbounded_channel();
        let mut filter = ActionFilter::new("test", true, false);
        let mut saw_fail = false;
        process_line(
            "openjd_env: MY_VAR=my_value",
            &mut filter,
            "test",
            &tx,
            &mut saw_fail,
        );
        match rx.try_recv().unwrap() {
            ActionMessage::SetEnv { name, value } => {
                assert_eq!(name, "MY_VAR");
                assert_eq!(value, "my_value");
            }
            other => panic!("Expected SetEnv, got: {other:?}"),
        }
    }

    #[test]
    fn test_process_line_unset_env() {
        let (tx, mut rx) = mpsc::unbounded_channel();
        let mut filter = ActionFilter::new("test", true, false);
        let mut saw_fail = false;
        process_line(
            "openjd_unset_env: MY_VAR",
            &mut filter,
            "test",
            &tx,
            &mut saw_fail,
        );
        match rx.try_recv().unwrap() {
            ActionMessage::UnsetEnv { name } => assert_eq!(name, "MY_VAR"),
            other => panic!("Expected UnsetEnv, got: {other:?}"),
        }
    }

    #[test]
    fn test_process_line_redacted_env() {
        let (tx, mut rx) = mpsc::unbounded_channel();
        let mut filter = ActionFilter::new("test", true, false);
        let mut saw_fail = false;
        process_line(
            "openjd_redacted_env: SECRET=hunter2",
            &mut filter,
            "test",
            &tx,
            &mut saw_fail,
        );
        match rx.try_recv().unwrap() {
            ActionMessage::RedactedEnv { name, value } => {
                assert_eq!(name, "SECRET");
                assert_eq!(value, "hunter2");
            }
            other => panic!("Expected RedactedEnv, got: {other:?}"),
        }
    }

    // ── Tier 2: Same-user integration tests ──────────────────────────

    #[cfg(unix)]
    fn run_simple(args: Vec<String>) -> (SubprocessResult, Vec<ActionMessage>) {
        run_with_config(SubprocessConfig {
            args,
            env_vars: HashMap::new(),
            working_dir: None,
            timeout: None,
            user: None,
            cancel_method: CancelMethod::Terminate,
            cancel_request_rx: None,
            debug_collect_stdout: true,
        })
    }

    #[cfg(unix)]
    fn run_with_config(config: SubprocessConfig) -> (SubprocessResult, Vec<ActionMessage>) {
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        rt.block_on(async {
            let (msg_tx, mut msg_rx) = mpsc::unbounded_channel();
            let mut filter = ActionFilter::new("test", true, false);
            let token = CancellationToken::new();
            let result = run_subprocess(config, &mut filter, "test", msg_tx, token)
                .await
                .unwrap();
            let mut msgs = Vec::new();
            while let Ok(m) = msg_rx.try_recv() {
                msgs.push(m);
            }
            (result, msgs)
        })
    }

    #[cfg(unix)]
    #[test]
    fn test_run_subprocess_success() {
        let (r, _) = run_simple(vec!["echo".into(), "hello".into()]);
        assert_eq!(r.state, ActionState::Success);
        assert_eq!(r.exit_code, Some(0));
        assert!(r.stdout.contains("hello"), "stdout: {}", r.stdout);
    }

    #[cfg(unix)]
    #[test]
    fn test_run_subprocess_failure_exit_code() {
        let (r, _) = run_simple(vec!["sh".into(), "-c".into(), "exit 42".into()]);
        assert_eq!(r.state, ActionState::Failed);
        assert_eq!(r.exit_code, Some(42));
    }

    #[cfg(unix)]
    #[test]
    fn test_run_subprocess_command_not_found() {
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let err = rt.block_on(async {
            let (msg_tx, _) = mpsc::unbounded_channel();
            let mut filter = ActionFilter::new("test", true, false);
            let token = CancellationToken::new();
            let config = SubprocessConfig {
                args: vec!["/nonexistent/binary_xyz".into()],
                env_vars: HashMap::new(),
                working_dir: None,
                timeout: None,
                user: None,
                cancel_method: CancelMethod::Terminate,
                cancel_request_rx: None,
                debug_collect_stdout: false,
            };
            run_subprocess(config, &mut filter, "test", msg_tx, token).await
        });
        assert!(err.is_err());
        let msg = err.unwrap_err().to_string();
        assert!(msg.contains("/nonexistent/binary_xyz"), "error: {msg}");
    }

    #[cfg(unix)]
    #[test]
    fn test_run_subprocess_empty_args() {
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let err = rt.block_on(async {
            let (msg_tx, _) = mpsc::unbounded_channel();
            let mut filter = ActionFilter::new("test", true, false);
            let token = CancellationToken::new();
            let config = SubprocessConfig {
                args: vec![],
                env_vars: HashMap::new(),
                working_dir: None,
                timeout: None,
                user: None,
                cancel_method: CancelMethod::Terminate,
                cancel_request_rx: None,
                debug_collect_stdout: false,
            };
            run_subprocess(config, &mut filter, "test", msg_tx, token).await
        });
        assert!(err.is_err());
        assert!(
            err.unwrap_err().to_string().contains("No command"),
            "expected empty args error"
        );
    }

    #[cfg(unix)]
    #[test]
    fn test_run_subprocess_timeout() {
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let (r, _) = rt.block_on(async {
            let (msg_tx, mut msg_rx) = mpsc::unbounded_channel();
            let mut filter = ActionFilter::new("test", true, false);
            let token = CancellationToken::new();
            let config = SubprocessConfig {
                args: vec!["sleep".into(), "30".into()],
                env_vars: HashMap::new(),
                working_dir: None,
                timeout: Some(Duration::from_millis(500)),
                user: None,
                cancel_method: CancelMethod::Terminate,
                cancel_request_rx: None,
                debug_collect_stdout: false,
            };
            let r = run_subprocess(config, &mut filter, "test", msg_tx, token)
                .await
                .unwrap();
            let mut msgs = Vec::new();
            while let Ok(m) = msg_rx.try_recv() {
                msgs.push(m);
            }
            (r, msgs)
        });
        assert_eq!(r.state, ActionState::Timeout);
    }

    #[cfg(unix)]
    #[test]
    fn test_run_subprocess_timeout_drains_stdout() {
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let (r, _) = rt.block_on(async {
            let (msg_tx, mut msg_rx) = mpsc::unbounded_channel();
            let mut filter = ActionFilter::new("test", true, false);
            let token = CancellationToken::new();
            let config = SubprocessConfig {
                args: vec![
                    "sh".into(),
                    "-c".into(),
                    "echo before_timeout; sleep 30".into(),
                ],
                env_vars: HashMap::new(),
                working_dir: None,
                timeout: Some(Duration::from_millis(500)),
                user: None,
                cancel_method: CancelMethod::Terminate,
                cancel_request_rx: None,
                debug_collect_stdout: true,
            };
            let r = run_subprocess(config, &mut filter, "test", msg_tx, token)
                .await
                .unwrap();
            let mut msgs = Vec::new();
            while let Ok(m) = msg_rx.try_recv() {
                msgs.push(m);
            }
            (r, msgs)
        });
        assert_eq!(r.state, ActionState::Timeout);
        assert!(
            r.stdout.contains("before_timeout"),
            "output before timeout should be captured: {:?}",
            r.stdout
        );
    }

    #[cfg(unix)]
    #[test]
    fn test_run_subprocess_env_vars() {
        let mut env = HashMap::new();
        env.insert("OPENJD_TEST_VAR".into(), Some("test_value_42".into()));
        let (r, _) = run_with_config(SubprocessConfig {
            args: vec!["sh".into(), "-c".into(), "echo $OPENJD_TEST_VAR".into()],
            env_vars: env,
            working_dir: None,
            timeout: None,
            user: None,
            cancel_method: CancelMethod::Terminate,
            cancel_request_rx: None,
            debug_collect_stdout: true,
        });
        assert_eq!(r.state, ActionState::Success);
        assert!(r.stdout.contains("test_value_42"), "stdout: {}", r.stdout);
    }

    #[cfg(unix)]
    #[test]
    fn test_run_subprocess_env_var_unset() {
        // Set a var then unset it — should not appear in child
        std::env::set_var("OPENJD_UNSET_TEST", "should_be_gone");
        let mut env = HashMap::new();
        env.insert("OPENJD_UNSET_TEST".into(), None);
        let (r, _) = run_with_config(SubprocessConfig {
            args: vec![
                "sh".into(),
                "-c".into(),
                "echo VAL=${OPENJD_UNSET_TEST:-UNSET}".into(),
            ],
            env_vars: env,
            working_dir: None,
            timeout: None,
            user: None,
            cancel_method: CancelMethod::Terminate,
            cancel_request_rx: None,
            debug_collect_stdout: true,
        });
        assert_eq!(r.state, ActionState::Success);
        assert!(r.stdout.contains("VAL=UNSET"), "stdout: {}", r.stdout);
    }

    #[cfg(unix)]
    #[test]
    fn test_run_subprocess_working_dir() {
        let dir = tempfile::tempdir().unwrap();
        let (r, _) = run_with_config(SubprocessConfig {
            args: vec!["pwd".into()],
            env_vars: HashMap::new(),
            working_dir: Some(dir.path().to_path_buf()),
            timeout: None,
            user: None,
            cancel_method: CancelMethod::Terminate,
            cancel_request_rx: None,
            debug_collect_stdout: true,
        });
        assert_eq!(r.state, ActionState::Success);
        // Resolve symlinks for comparison (macOS /tmp -> /private/tmp)
        let expected = dir.path().canonicalize().unwrap();
        let actual = PathBuf::from(r.stdout.trim()).canonicalize().unwrap();
        assert_eq!(actual, expected);
    }

    #[cfg(unix)]
    #[test]
    fn test_run_subprocess_openjd_progress() {
        let (r, msgs) = run_simple(vec![
            "sh".into(),
            "-c".into(),
            "echo 'openjd_progress: 0.75'".into(),
        ]);
        assert_eq!(r.state, ActionState::Success);
        assert!(
            msgs.iter().any(
                |m| matches!(m, ActionMessage::Progress(v) if (*v - 0.75).abs() < f64::EPSILON)
            ),
            "Expected Progress(0.75), got: {msgs:?}"
        );
    }

    #[cfg(unix)]
    #[test]
    fn test_run_subprocess_openjd_status() {
        let (r, msgs) = run_simple(vec![
            "sh".into(),
            "-c".into(),
            "echo 'openjd_status: rendering'".into(),
        ]);
        assert_eq!(r.state, ActionState::Success);
        assert!(
            msgs.iter()
                .any(|m| matches!(m, ActionMessage::Status(s) if s == "rendering")),
            "Expected Status(rendering), got: {msgs:?}"
        );
    }

    #[cfg(unix)]
    #[test]
    fn test_run_subprocess_openjd_fail_sets_failed() {
        let (r, msgs) = run_simple(vec![
            "sh".into(),
            "-c".into(),
            "echo 'openjd_fail: something broke'".into(),
        ]);
        assert_eq!(
            r.state,
            ActionState::Failed,
            "openjd_fail should cause Failed state even with exit 0"
        );
        assert!(
            msgs.iter()
                .any(|m| matches!(m, ActionMessage::Fail(s) if s == "something broke")),
            "Expected Fail message, got: {msgs:?}"
        );
    }

    #[cfg(unix)]
    #[test]
    fn test_run_subprocess_openjd_env() {
        let (r, msgs) = run_simple(vec![
            "sh".into(),
            "-c".into(),
            "echo 'openjd_env: FOO=bar'".into(),
        ]);
        assert_eq!(r.state, ActionState::Success);
        assert!(msgs.iter().any(|m| matches!(m, ActionMessage::SetEnv { name, value } if name == "FOO" && value == "bar")),
            "Expected SetEnv, got: {msgs:?}");
    }

    #[cfg(unix)]
    #[test]
    fn test_run_subprocess_stderr_merged() {
        // stderr should be merged into stdout
        let (r, _) = run_simple(vec![
            "sh".into(),
            "-c".into(),
            "echo stdout_line; echo stderr_line >&2".into(),
        ]);
        assert_eq!(r.state, ActionState::Success);
        assert!(r.stdout.contains("stdout_line"), "stdout: {}", r.stdout);
        assert!(
            r.stdout.contains("stderr_line"),
            "stderr should be merged into stdout: {}",
            r.stdout
        );
    }

    #[cfg(unix)]
    #[test]
    fn test_run_subprocess_multiline_output() {
        let (r, _) = run_simple(vec![
            "sh".into(),
            "-c".into(),
            "echo line1; echo line2; echo line3".into(),
        ]);
        assert_eq!(r.state, ActionState::Success);
        assert!(r.stdout.contains("line1\n"), "stdout: {:?}", r.stdout);
        assert!(r.stdout.contains("line2\n"), "stdout: {:?}", r.stdout);
        assert!(r.stdout.contains("line3\n"), "stdout: {:?}", r.stdout);
    }

    #[cfg(unix)]
    #[test]
    fn test_run_subprocess_debug_collect_stdout_false_by_default() {
        let (r, _) = run_simple(vec!["echo".into(), "hello".into()]);
        // run_simple sets debug_collect_stdout: true, so stdout is captured
        assert!(r.stdout.contains("hello"));

        // With debug_collect_stdout: false (default), stdout should be empty
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let r = rt.block_on(async {
            let (msg_tx, _) = mpsc::unbounded_channel();
            let mut filter = ActionFilter::new("test", true, false);
            let token = CancellationToken::new();
            let config = SubprocessConfig {
                args: vec!["echo".into(), "hello".into()],
                env_vars: HashMap::new(),
                working_dir: None,
                timeout: None,
                user: None,
                cancel_method: CancelMethod::Terminate,
                cancel_request_rx: None,
                debug_collect_stdout: false,
            };
            run_subprocess(config, &mut filter, "test", msg_tx, token)
                .await
                .unwrap()
        });
        assert_eq!(r.state, ActionState::Success);
        assert!(
            r.stdout.is_empty(),
            "stdout should be empty when debug_collect_stdout is false: {:?}",
            r.stdout
        );
    }

    #[test]
    fn test_truncate_line_multibyte_boundary() {
        // '€' is 3 bytes in UTF-8. LOG_LINE_MAX_LENGTH (65536) % 3 == 1,
        // so the byte boundary falls inside a multi-byte character.
        let s = "€".repeat(LOG_LINE_MAX_LENGTH); // 3 * LOG_LINE_MAX_LENGTH bytes
        let truncated = truncate_line(&s);
        assert!(truncated.len() <= LOG_LINE_MAX_LENGTH);
        // Must be valid UTF-8 (the fact that we can call .chars() without panic proves it)
        assert!(truncated.chars().count() > 0);
    }

    #[test]
    fn test_truncate_line_short_line_unchanged() {
        let s = "hello";
        assert_eq!(truncate_line(s), "hello");
    }

    /// Conformance table for [`decode_backslashreplace`].
    ///
    /// Every expected value was generated by CPython's
    /// `bytes.decode("utf-8", errors="backslashreplace")`, which is what
    /// `openjd-sessions-for-python` uses. These pin byte-for-byte parity with
    /// the Python implementation, including how many escapes a multi-byte
    /// invalid sequence produces (CPython escapes each byte individually).
    #[test]
    fn test_decode_backslashreplace_matches_cpython() {
        let cases: &[(&[u8], &str)] = &[
            // The customer's byte: 0x97 is the cp1252 em dash (Unreal Engine on Windows).
            (
                &[0x62, 0x61, 0x64, 0x20, 0x97, 0x20, 0x62, 0x79, 0x74, 0x65],
                r"bad \x97 byte",
            ),
            // 0xff is never valid anywhere in UTF-8.
            (
                &[0x62, 0x61, 0x64, 0x20, 0xff, 0x20, 0x62, 0x79, 0x74, 0x65],
                r"bad \xff byte",
            ),
            // Consecutive invalid bytes are escaped separately.
            (
                &[
                    0x62, 0x61, 0x64, 0x20, 0xc7, 0xff, 0x20, 0x62, 0x79, 0x74, 0x65, 0x73,
                ],
                r"bad \xc7\xff bytes",
            ),
            // A cp1252 text run ("Çé"), invalid as UTF-8.
            (
                &[
                    0x62, 0x61, 0x64, 0x20, 0xc7, 0xe9, 0x20, 0x74, 0x65, 0x78, 0x74,
                ],
                r"bad \xc7\xe9 text",
            ),
            // Truncated 3-byte sequence followed by valid ASCII.
            (
                &[
                    0x74, 0x72, 0x75, 0x6e, 0x63, 0x61, 0x74, 0x65, 0x64, 0x20, 0xe4, 0xbd, 0x20,
                    0x74, 0x68, 0x65, 0x6e, 0x20, 0x6f, 0x6b,
                ],
                r"truncated \xe4\xbd then ok",
            ),
            // Truncated sequence at end of input: `Utf8Error::error_len()` is
            // `None` here, so this pins the unterminated-sequence path.
            (
                &[0x74, 0x61, 0x69, 0x6c, 0x20, 0xe4, 0xbd],
                r"tail \xe4\xbd",
            ),
            // A lone continuation byte with no lead byte.
            (
                &[0x80, 0x20, 0x6c, 0x65, 0x61, 0x64, 0x69, 0x6e, 0x67],
                r"\x80 leading",
            ),
            // Valid 2- and 3-byte sequences pass through unmodified.
            (
                &[
                    0x68, 0xc3, 0xa9, 0x6c, 0x6c, 0x6f, 0x20, 0x77, 0xc3, 0xb6, 0x72, 0x6c, 0x64,
                    0x20, 0xc3, 0x87, 0x20, 0xe6, 0x98, 0x9f, 0xe6, 0x9c, 0x9f, 0xe4, 0xba, 0x94,
                ],
                "h\u{e9}llo w\u{f6}rld \u{c7} \u{661f}\u{671f}\u{4e94}",
            ),
            // Valid 4-byte sequence (emoji) passes through unmodified.
            (
                &[
                    0x6f, 0x6b, 0x20, 0xf0, 0x9f, 0x98, 0x80, 0x20, 0x64, 0x6f, 0x6e, 0x65,
                ],
                "ok \u{1f600} done",
            ),
            // A UTF-8-encoded surrogate half: all three bytes are escaped.
            (
                &[0x73, 0x20, 0xed, 0xa0, 0x80, 0x20, 0x65],
                r"s \xed\xa0\x80 e",
            ),
            // An overlong encoding of '/': both bytes are escaped.
            (&[0x6f, 0x20, 0xc0, 0xaf, 0x20, 0x65], r"o \xc0\xaf e"),
            // A lead byte beyond the Unicode maximum: all four bytes are escaped.
            (
                &[0x72, 0x20, 0xf5, 0x80, 0x80, 0x80, 0x20, 0x65],
                r"r \xf5\x80\x80\x80 e",
            ),
            (&[], ""),
            (
                &[
                    0x70, 0x6c, 0x61, 0x69, 0x6e, 0x20, 0x61, 0x73, 0x63, 0x69, 0x69,
                ],
                "plain ascii",
            ),
            // Invalid bytes at both edges of the input.
            (&[0xff, 0x6d, 0x69, 0x64, 0xfe], r"\xffmid\xfe"),
            // A backslash already in the output is not doubled: only
            // undecodable bytes are escaped.
            (
                &[
                    0x43, 0x3a, 0x5c, 0x70, 0x61, 0x74, 0x68, 0x5c, 0x78, 0x34, 0x31, 0x20, 0x97,
                ],
                r"C:\path\x41 \x97",
            ),
            // NUL and tab are valid UTF-8 and pass through unescaped.
            (&[0x61, 0x00, 0x62, 0x09, 0x63], "a\0b\tc"),
        ];

        for (input, expected) in cases {
            assert_eq!(
                decode_backslashreplace(input),
                *expected,
                "input bytes: {input:02x?}"
            );
        }
    }

    #[test]
    fn test_decode_backslashreplace_borrows_valid_input() {
        // Fully valid input must not allocate: the common case is a hot path
        // running once per line of subprocess output.
        assert!(matches!(
            decode_backslashreplace("valid \u{661f} text".as_bytes()),
            Cow::Borrowed(_)
        ));
        assert!(matches!(
            decode_backslashreplace(&[0x62, 0x61, 0x64, 0x20, 0x97]),
            Cow::Owned(_)
        ));
    }

    #[test]
    fn test_decode_backslashreplace_escapes_use_lowercase_hex() {
        // CPython emits lowercase hex; an uppercase escape would be a visible
        // divergence in the logs.
        assert_eq!(decode_backslashreplace(&[0xab, 0xcd]), r"\xab\xcd");
    }

    #[cfg(unix)]
    #[test]
    fn test_run_subprocess_invalid_utf8_continues() {
        // printf outputs raw bytes: valid line, then 0xFF (invalid UTF-8), then another valid line.
        // All lines including those after invalid bytes must be captured.
        let (r, _) = run_simple(vec![
            "sh".into(),
            "-c".into(),
            r#"echo before; printf '\xff\n'; echo after"#.into(),
        ]);
        assert_eq!(r.state, ActionState::Success);
        assert!(
            r.stdout.contains("before"),
            "line before invalid UTF-8 should be captured: {:?}",
            r.stdout
        );
        assert!(
            r.stdout.contains("after"),
            "line after invalid UTF-8 should be captured: {:?}",
            r.stdout
        );
    }

    #[cfg(unix)]
    #[test]
    fn test_run_subprocess_invalid_utf8_is_escaped() {
        // Undecodable bytes in subprocess output must be escaped as `\xNN`,
        // preserving the original byte values in the session log rather than
        // collapsing them to U+FFFD. 0x97 is the cp1252 em dash, the byte a
        // customer's Unreal Engine renderer emitted on Windows.
        let (r, _) = run_simple(vec![
            "sh".into(),
            "-c".into(),
            r#"printf 'bad \x97 byte!\n'"#.into(),
        ]);
        assert_eq!(r.state, ActionState::Success);
        assert!(
            r.stdout.contains(r"bad \x97 byte!"),
            "undecodable byte should be escaped as \\x97, preserving its value: {:?}",
            r.stdout
        );
        assert!(
            !r.stdout.contains('\u{fffd}'),
            "the replacement character must not appear; the byte value must be preserved: {:?}",
            r.stdout
        );
    }

    #[cfg(unix)]
    #[test]
    fn test_run_subprocess_valid_utf8_not_escaped() {
        // Valid multi-byte UTF-8 must pass through unmodified. Negative control
        // against over-escaping.
        let (r, _) = run_simple(vec![
            "sh".into(),
            "-c".into(),
            "printf 'h\u{e9}llo w\u{f6}rld \u{661f}\u{671f}\u{4e94}\\n'".into(),
        ]);
        assert_eq!(r.state, ActionState::Success);
        assert!(
            r.stdout
                .contains("h\u{e9}llo w\u{f6}rld \u{661f}\u{671f}\u{4e94}"),
            "valid UTF-8 should pass through unmodified: {:?}",
            r.stdout
        );
        assert!(
            !r.stdout.contains(r"\x"),
            "valid UTF-8 must not be escaped: {:?}",
            r.stdout
        );
    }

    #[cfg(unix)]
    #[test]
    fn test_run_subprocess_progress_error_in_stdout() {
        let (r, _) = run_simple(vec![
            "sh".into(),
            "-c".into(),
            "echo 'openjd_progress: 200.0'".into(),
        ]);
        assert!(
            r.stdout.contains("ERROR"),
            "out-of-range progress error should appear in stdout: {:?}",
            r.stdout
        );
    }

    /// When a process exits with non-zero and the cancel token has been
    /// cancelled, the result should be `Canceled` not `Failed`.
    ///
    /// This covers the pyo3 binding's cancel path where the cross-user helper
    /// kills the process (non-zero exit) while the token is cancelled, but the
    /// select loop's cancel branch may not have fired (so `cancel_requested`
    /// could be false). The `is_cancelled()` check in state determination
    /// ensures the correct result regardless of select ordering.
    #[cfg(unix)]
    #[tokio::test]
    async fn test_cancel_token_set_but_process_killed_externally() {
        use tokio_util::sync::CancellationToken;

        let token = CancellationToken::new();
        let (msg_tx, _msg_rx) = tokio::sync::mpsc::unbounded_channel();

        // Process that exits immediately with non-zero
        let config = SubprocessConfig {
            args: vec!["sh".into(), "-c".into(), "exit 42".into()],
            env_vars: HashMap::new(),
            working_dir: None,
            timeout: None,
            user: None,
            cancel_method: CancelMethod::Terminate,
            cancel_request_rx: None,
            debug_collect_stdout: false,
        };

        // Cancel from OS thread — simulates the pyo3 binding's cancel path
        let token_clone = token.clone();
        std::thread::spawn(move || {
            token_clone.cancel();
        });

        // Small yield to let the OS thread run
        tokio::time::sleep(Duration::from_millis(1)).await;

        let mut filter = crate::action_filter::ActionFilter::new("test", true, false);
        let result = run_subprocess(config, &mut filter, "test", msg_tx, token)
            .await
            .unwrap();

        // The token is cancelled and the process exited non-zero.
        // The result must be Canceled, not Failed.
        assert_eq!(
            result.state,
            ActionState::Canceled,
            "Non-zero exit with cancelled token should be Canceled, not {:?}",
            result.state
        );
    }
}

#[cfg(test)]
mod collect_tree_validated_tests {
    use super::*;

    /// Build a `creation_time` callback from a slice of `(pid, ctime)` pairs.
    /// Any pid absent from the slice reports an unreadable creation time.
    fn ctimes_from(pairs: &[(u32, u64)]) -> impl FnMut(u32) -> Option<u64> + '_ {
        move |pid| pairs.iter().find(|(p, _)| *p == pid).map(|(_, ct)| *ct)
    }

    fn vp(pid: u32, creation_time: u64) -> ValidatedProcess {
        ValidatedProcess { pid, creation_time }
    }

    #[test]
    fn collects_true_descendants_in_bfs_order() {
        let parents = [(20u32, 10u32), (30, 20)];
        let mut ct = ctimes_from(&[(10, 100), (20, 200), (30, 300)]);
        let got = collect_tree_validated(10, &parents, u64::MAX, None, &mut ct);
        assert_eq!(got, vec![vp(10, 100), vp(20, 200), vp(30, 300)]);
    }

    #[test]
    fn rejects_stale_edge_child_older_than_parent() {
        let parents = [(40u32, 10u32)];
        let mut ct = ctimes_from(&[(10, 500), (40, 100)]);
        let got = collect_tree_validated(10, &parents, u64::MAX, None, &mut ct);
        assert_eq!(got, vec![vp(10, 500)]);
    }

    #[test]
    fn unreadable_child_stays_traversable_with_parent_bound() {
        // A child whose creation time is unreadable under a validated parent is
        // no longer pruned: it is recorded carrying the parent's creation time
        // so its own descendants remain reachable. kill_process_checked will
        // no-op on the unreadable node at kill time.
        let parents = [(20u32, 10u32)];
        let mut ct = ctimes_from(&[(10, 100)]);
        let got = collect_tree_validated(10, &parents, u64::MAX, None, &mut ct);
        assert_eq!(got, vec![vp(10, 100), vp(20, 100)]);
    }

    #[test]
    fn unreadable_intermediate_keeps_live_grandchild() {
        // cmd(10) -> sh(20, exited/unreadable) -> renderer(30, live). The dead
        // intermediate must stay traversable carrying its parent's bound (100)
        // so the live grandchild is still enumerated and validated.
        let parents = [(20u32, 10u32), (30, 20)];
        let mut ct = ctimes_from(&[(10, 100), (30, 300)]);
        let got = collect_tree_validated(10, &parents, u64::MAX, None, &mut ct);
        assert_eq!(got, vec![vp(10, 100), vp(20, 100), vp(30, 300)]);
    }

    #[test]
    fn unreadable_intermediate_bound_still_prunes_stale_grandchild() {
        // Same shape, but the grandchild reads OLDER than the inherited parent
        // bound (50 < 100): it is a stale reused-parent edge and is rejected.
        let parents = [(20u32, 10u32), (30, 20)];
        let mut ct = ctimes_from(&[(10, 100), (30, 50)]);
        let got = collect_tree_validated(10, &parents, u64::MAX, None, &mut ct);
        assert_eq!(got, vec![vp(10, 100), vp(20, 100)]);
    }

    #[test]
    fn dead_root_with_pinned_identity_reaps_orphans() {
        // Root(10) exited during the grace period (unreadable) but a
        // schedule-time identity (200) was pinned. Seed the walk with the pin
        // and reap the orphaned descendant(20).
        let parents = [(20u32, 10u32)];
        let mut ct = ctimes_from(&[(20, 250)]);
        let got = collect_tree_validated(10, &parents, u64::MAX, Some(200), &mut ct);
        assert_eq!(got, vec![vp(10, 200), vp(20, 250)]);
    }

    #[test]
    fn returns_empty_when_root_ctime_unreadable() {
        // Unpinned (expected_root_ct = None) root that is unreadable: nothing to
        // validate against, so fail closed. Contrast dead_root_with_pinned_
        // identity_reaps_orphans, where a pinned identity seeds the walk.
        let parents = [(20u32, 10u32)];
        let mut ct = ctimes_from(&[(20, 200)]);
        let got = collect_tree_validated(10, &parents, u64::MAX, None, &mut ct);
        assert_eq!(got, Vec::<ValidatedProcess>::new());
    }

    #[test]
    fn stale_cycle_terminates() {
        let parents = [(20u32, 10u32), (10, 20)];
        let mut ct = ctimes_from(&[(10, 500), (20, 100)]);
        let got = collect_tree_validated(10, &parents, u64::MAX, None, &mut ct);
        assert_eq!(got, vec![vp(10, 500)]);
    }

    #[test]
    fn accepts_equal_creation_time() {
        let parents = [(20u32, 10u32)];
        let mut ct = ctimes_from(&[(10, 100), (20, 100)]);
        let got = collect_tree_validated(10, &parents, u64::MAX, None, &mut ct);
        assert_eq!(got, vec![vp(10, 100), vp(20, 100)]);
    }

    #[test]
    fn prunes_entire_subtree_below_rejected_edge() {
        let parents = [(20u32, 10u32), (30, 20)];
        let mut ct = ctimes_from(&[(10, 500), (20, 100), (30, 600)]);
        let got = collect_tree_validated(10, &parents, u64::MAX, None, &mut ct);
        assert_eq!(got, vec![vp(10, 500)]);
    }

    #[test]
    fn equal_time_cycle_terminates_without_duplicates() {
        let parents = [(20u32, 10u32), (10, 20)];
        let mut ct = ctimes_from(&[(10, 100), (20, 100)]);
        let got = collect_tree_validated(10, &parents, u64::MAX, None, &mut ct);
        assert_eq!(got, vec![vp(10, 100), vp(20, 100)]);
    }

    #[test]
    fn identity_match_rejects_changed_creation_time() {
        assert!(!process_identity_matches(200, Some(900)));
        assert!(!process_identity_matches(200, None));
    }

    #[test]
    fn identity_match_accepts_same_creation_time() {
        assert!(process_identity_matches(200, Some(200)));
    }

    #[test]
    fn rejects_candidate_created_after_snapshot() {
        // Finding 1: the edge (20,10) is genuine at snapshot time, but pid 20
        // exited and was reused by a newer process. snapshot_time bounds out
        // the impostor even though ctime 500 passes the `>=` parent check.
        let parents = [(20u32, 10u32)];
        let mut ct = ctimes_from(&[(10, 100), (20, 500)]);
        let got = collect_tree_validated(10, &parents, 300, None, &mut ct);
        assert_eq!(got, vec![vp(10, 100)]);
    }

    #[test]
    fn immediate_root_newer_than_snapshot_still_collected() {
        // Immediate send_terminate path (expected None): the live Child handle
        // pins the PID, so the root is exempt from the snapshot bound even
        // though its ctime (900) post-dates the stamp (300). A descendant that
        // also post-dates the stamp (950) is still rejected.
        let parents = [(20u32, 10u32)];
        let mut ct = ctimes_from(&[(10, 900), (20, 950)]);
        let got = collect_tree_validated(10, &parents, 300, None, &mut ct);
        assert_eq!(got, vec![vp(10, 900)]);
    }

    #[test]
    fn pinned_root_newer_than_snapshot_still_collected() {
        // Delayed path with a matching pin: the identity match proves the
        // pinned process, so the root is collected despite ctime (900) > stamp
        // (300).
        let parents: [(u32, u32); 0] = [];
        let mut ct = ctimes_from(&[(10, 900)]);
        let got = collect_tree_validated(10, &parents, 300, Some(900), &mut ct);
        assert_eq!(got, vec![vp(10, 900)]);
    }

    #[test]
    fn root_exemption_does_not_extend_to_children() {
        // The root (100) is exempt from the stamp (300), but children still
        // obey it: 250 is accepted, 400 is rejected as a post-snapshot reuse.
        let parents = [(20u32, 10u32), (30, 10)];
        let mut ct = ctimes_from(&[(10, 100), (20, 250), (30, 400)]);
        let got = collect_tree_validated(10, &parents, 300, None, &mut ct);
        assert_eq!(got, vec![vp(10, 100), vp(20, 250)]);
    }

    #[test]
    fn root_expected_ctime_mismatch_returns_empty() {
        // Finding 2: pinned schedule-time identity 200, but the root now reads
        // 900 => reused or exited => fail closed.
        let parents = [(20u32, 10u32)];
        let mut ct = ctimes_from(&[(10, 900), (20, 950)]);
        let got = collect_tree_validated(10, &parents, u64::MAX, Some(200), &mut ct);
        assert_eq!(got, Vec::<ValidatedProcess>::new());
    }

    #[test]
    fn root_expected_ctime_match_uses_expected_identity() {
        // Pinned identity 200 matches the fresh read; the recorded root carries
        // the pinned value and children are collected normally.
        let parents = [(20u32, 10u32), (30, 20)];
        let mut ct = ctimes_from(&[(10, 200), (20, 300), (30, 400)]);
        let got = collect_tree_validated(10, &parents, u64::MAX, Some(200), &mut ct);
        assert_eq!(got, vec![vp(10, 200), vp(20, 300), vp(30, 400)]);
        assert_eq!(got[0].creation_time, 200);
    }
}