term-wm-pty-engine 0.9.24-alpha

use std::io::{Read, Write};
use std::sync::{
    Arc, Condvar, Mutex, OnceLock,
    atomic::{AtomicBool, AtomicUsize, Ordering},
};
use std::thread::{self, JoinHandle};
use std::time::Instant;

use portable_pty::{Child, CommandBuilder, MasterPty, PtySize, native_pty_system};

#[cfg(windows)]
use crate::job_object::JobObject;
use crate::pty_state_tracker::PtyPerformAdapter;
use term_clipboard::{Clipboard, Osc52Extractor};

/// Size of the PTY master read buffer (single `read()` call).
/// 64KB keeps the reader parked most of the time under heavy output
/// (64KB × 60fps ≈ 3.8MB/s throughput, enough for any terminal workload).
const PTY_READ_BUF_SIZE: usize = 65536;

/// Drain-starvation bound (Unix only): a continuous stream (`yes`, busy logs)
/// never returns WouldBlock, so apply any pending resize at least this often to
/// keep the grid attached to the physical window bounds. On Windows the reader
/// applies pending resizes at every (blocking) read boundary instead.
#[cfg(unix)]
const MAX_DRAIN_BYTES: usize = 65536;

/// Windows error `ERROR_OPERATION_ABORTED` (995): the code a blocking
/// `ReadFile` returns when `CancelSynchronousIo` aborts it. The reader treats
/// this as a resize wake — never a fatal read error. (Inert on Unix, where
/// `errno` can never be 995.)
const ERROR_OPERATION_ABORTED: i32 = 995;

#[cfg(unix)]
use std::os::unix::io::RawFd;

/// Wake primitive that nudges the reader thread out of `poll` when the UI
/// requests a resize (Unix). A self-pipe: the read end is polled alongside the
/// PTY master fd, and `signal` writes a byte. On Windows there is no pipe to
/// poll — the ConPTY output pipe is a blocking anonymous pipe — so the wake is
/// instead delivered by `CancelSynchronousIo` in [`Pty::wake_reader`].
#[cfg(unix)]
struct ResizeWake {
    read_fd: RawFd,
    write_fd: RawFd,
}

#[cfg(unix)]
impl ResizeWake {
    fn new() -> std::io::Result<Self> {
        let mut fds = [0i32; 2];
        // SAFETY: pipe writes to a valid 2-element array.
        if unsafe { libc::pipe(fds.as_mut_ptr()) } != 0 {
            return Err(std::io::Error::last_os_error());
        }
        // Both ends non-blocking so poll() returns immediately when signalled
        // and a signal never blocks.
        for fd in &fds {
            // SAFETY: fds are valid descriptors returned by pipe() above.
            unsafe { libc::fcntl(*fd, libc::F_SETFL, libc::O_NONBLOCK) };
        }
        Ok(Self {
            read_fd: fds[0],
            write_fd: fds[1],
        })
    }

    fn read_fd(&self) -> RawFd {
        self.read_fd
    }

    fn signal(&self) {
        let byte = [1u8];
        // SAFETY: write one byte to the pipe write end.
        unsafe {
            let _ = libc::write(self.write_fd, byte.as_ptr() as *const libc::c_void, 1);
        }
    }
}

#[cfg(unix)]
impl Drop for ResizeWake {
    fn drop(&mut self) {
        // SAFETY: close the two pipe ends created in `new`.
        unsafe {
            libc::close(self.read_fd);
            libc::close(self.write_fd);
        }
    }
}

/// Clear (drain) any pending wake bytes on the reader side of the self-pipe.
#[cfg(unix)]
fn clear_wake(fd: RawFd) {
    let mut buf = [0u8; 64];
    loop {
        // SAFETY: read into a valid buffer.
        let n = unsafe { libc::read(fd, buf.as_mut_ptr() as *mut libc::c_void, buf.len()) };
        if n <= 0 {
            break;
        }
    }
}

/// Number of bytes from the end of the previous chunk to carry forward
/// for cross-boundary pattern detection (DSR, OSC 52 header).
/// Must cover the 5-byte OSC 52 header `\x1b]52;` across chunk boundaries
/// (needs at least 4 bytes of tail; 8 provides margin).
const HISTORY_TAIL_LEN: usize = 8;

/// Length of the DSR request sequence `\x1b[6n`.
const DSR_PATTERN_LEN: usize = 4;

/// Env var that enables dumping raw PTY→emulator bytes (as hex) to a file.
/// Temporary diagnostic aid for seeing exactly what a child app sends (e.g.
/// pico's escape sequences at the right margin of a long line).
const ESC_TRACE_ENV: &str = "TERM_WM_TRACE_ESC";

/// Whether `ESC_TRACE_ENV` is set — checked once per process.
static ESC_TRACE_ENABLED: OnceLock<bool> = OnceLock::new();

/// Append a chunk of raw bytes fed to the emulator as a hex line to the file
/// named by `TERM_WM_TRACE_ESC` (default `term_wm_esc_trace.log`). Off by
/// default; no-op when the env var is unset.
fn esc_trace_chunk(bytes: &[u8]) {
    if !*ESC_TRACE_ENABLED.get_or_init(|| std::env::var_os(ESC_TRACE_ENV).is_some()) {
        return;
    }
    let path = std::env::var(ESC_TRACE_ENV).unwrap_or_else(|_| "term_wm_esc_trace.log".to_string());
    if let Ok(mut file) = std::fs::OpenOptions::new()
        .create(true)
        .append(true)
        .open(&path)
    {
        let _ = writeln!(file, "{}", hex_bytes(bytes));
    }
}

/// Format bytes as a lowercase hex string (hand-rolled; no extra dependency).
fn hex_bytes(bytes: &[u8]) -> String {
    let mut out = String::with_capacity(bytes.len() * 2);
    for b in bytes {
        out.push_str(&format!("{b:02x}"));
    }
    out
}

/// Buffer size for `proc_name()` on macOS.
#[cfg(target_os = "macos")]
const PROC_NAME_BUF_SIZE: usize = 64;

/// How often to check the foreground process group for title changes.
const FOREGROUND_POLL_INTERVAL: std::time::Duration = std::time::Duration::from_secs(1);
use crate::PtyStatus;
use crate::title::extract_osc_title;

pub type PtyResult<T> = Result<T, Box<dyn std::error::Error + Send + Sync>>;

type StatusCallback = Arc<Mutex<Option<Box<dyn Fn(PtyStatus) + Send + Sync>>>>;

/// Cloneable handle to the PTY input writer.
///
/// Writes to a PTY/ConPTY master are blocking synchronous I/O: when the
/// kernel input buffer fills (fast typing, large paste, or a child that
/// paused reading stdin), `write_all` blocks the calling thread. Callers
/// running on an async runtime (e.g. the session server) must offload writes
/// to a blocking thread (tokio's `spawn_blocking`) via this handle, so a
/// full input buffer never starves a runtime worker. The internal mutex
/// serializes concurrent writers and keeps each `write_bytes` call atomic.
#[derive(Clone)]
pub struct PtyWriter {
    inner: Arc<Mutex<Box<dyn Write + Send>>>,
}

impl PtyWriter {
    fn new(writer: Box<dyn Write + Send>) -> Self {
        Self {
            inner: Arc::new(Mutex::new(writer)),
        }
    }

    /// Write bytes to the PTY master. Blocks until the bytes are accepted
    /// (or the writer fails); call from a dedicated blocking thread.
    pub fn write_bytes(&self, input: &[u8]) -> std::io::Result<()> {
        let mut writer = self.inner.lock().unwrap_or_else(|err| err.into_inner());
        writer.write_all(input)?;
        writer.flush()
    }
}

pub struct Pty {
    /// Shared master for Unix-only process-group queries (`foreground_pid`,
    /// `process_group_id`, `signal_process_group`). The reader thread owns its
    /// own clone for drain-synchronized resizes, so on Windows this field is
    /// absent entirely.
    #[cfg(unix)]
    master: Arc<Mutex<Box<dyn MasterPty + Send>>>,
    writer: PtyWriter,
    /// Raw bytes from the reader thread, kept for consumers that need
    /// unparsed output (e.g., session server forwarding).
    pending: Arc<Mutex<Vec<u8>>>,
    bytes_received: Arc<AtomicUsize>,
    last_bytes: Arc<Mutex<Vec<u8>>>,
    dsr_requested: Arc<AtomicBool>,
    pending_title: Arc<Mutex<Option<String>>>,
    foreground_title: Arc<Mutex<Option<String>>>,
    last_fg_pid: u32,
    last_fg_check: Instant,
    /// Parsed screen shared between the reader thread and the main thread.
    /// The reader parses bytes into this parser in-place. The main thread
    /// locks it to read cells directly — zero clones.
    pub(crate) shared_parser: Arc<Mutex<term_wm_vt100::Parser>>,
    /// Set by the reader thread when new content has been parsed.
    pub(crate) dirty: Arc<AtomicBool>,
    /// Condvar for I/O burst budget: reader waits here when budget exceeded
    /// and the UI hasn't rendered yet.
    pub(crate) dirty_cond: Arc<(Mutex<()>, Condvar)>,
    /// Current emulator size. Owned/shared: the reader applies drain-synchronized
    /// resizes and updates this; the main thread reads it via [`Pty::size`].
    size: Arc<Mutex<PtySize>>,
    /// Latest resize requested by the UI, applied by the reader thread at the
    /// next pipe-drain boundary (never mid-shell-write).
    pending_resize: Arc<Mutex<Option<PtySize>>>,
    /// Wake primitive used to nudge the reader out of `poll` on a resize request.
    #[cfg(unix)]
    resize_wake: ResizeWake,
    scrollback_len: usize,
    child: Option<Box<dyn Child + Send + Sync>>,
    /// Win32 Job Object containing the child's process tree (Windows only).
    /// Dropping it with `JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE` terminates any
    /// processes still in the job, so an un-killed `Pty` never orphans
    /// descendants. The Windows analogue of the Unix process group.
    #[cfg(windows)]
    job: Option<JobObject>,
    exited: bool,
    exit_status: Option<portable_pty::ExitStatus>,
    /// Guards against double-fire of the Exited callback from both
    /// has_exited() and the reader thread (EOF detection race).
    exited_emitted: Arc<AtomicBool>,
    reader: Option<JoinHandle<()>>,
    /// Status callback invoked by the reader thread on wakeup and exit.
    status_cb: StatusCallback,
    /// Application state tracker (alternate screen, mouse tracking, margins).
    /// Shared via Arc: reader thread writes (via PtyPerformAdapter), main thread reads.
    pub(crate) tracker: std::sync::Arc<crate::PtyStateTracker>,
    /// Shutdown flag: when true, the reader thread exits its loop ASAP.
    /// Set by into_parts() and Drop.
    shutdown: Arc<AtomicBool>,
}

/// The bounded channel between PTY reader threads and the main event loop
/// provides mechanical backpressure: when the channel is full, the reader
/// thread's `send()` blocks → the PTY master read call pauses → the OS
/// pipe buffer fills → the child process's `write()` blocks. This prevents
/// memory exhaustion when output floods faster than the UI can render.
/// Parts of a `Pty` that can be moved into the `Reaper` for async teardown.
pub struct PtyParts {
    pub child: Option<Box<dyn Child + Send + Sync>>,
    pub reader_handle: Option<JoinHandle<()>>,
}

/// Constant env values applied to every spawned PTY child so ncurses and
/// modern CLI tools render correctly when the session runs across SSH /
/// container hops (see `sanitize_child_environment`).
///
/// macOS ships a defective `screen-256color` terminfo lacking the `bce`
/// (Background Color Erase) capability, which makes ncurses apps (pico, nano)
/// reset attributes before line erases and drop background colors.
/// `xterm-256color` has `bce` and is safe on macOS now that the invalid
/// `LC_CTYPE=UTF-8` is stripped (the grid-math crashes previously associated
/// with it were caused by `setlocale` failing on that locale, not the terminfo
/// entry). Other platforms keep `screen-256color` — their terminfo has `bce`.
#[cfg(target_os = "macos")]
const CHILD_TERM: &str = "xterm-256color";
#[cfg(not(target_os = "macos"))]
const CHILD_TERM: &str = "screen-256color";
const CHILD_COLORTERM: &str = "truecolor";

/// Apply the multiplexer-safe environment to a spawned PTY child.
///
/// - Forces `TERM` to a terminfo entry with `bce` (Background Color Erase):
///   `xterm-256color` on macOS (where Apple's `screen-256color` terminfo is
///   broken and missing `bce`), `screen-256color` elsewhere. Without `bce`,
///   ncurses apps reset attributes before line erases, which erases the
///   background in the emulated screen.
/// - Opts into `COLORTERM=truecolor` for modern CLI tools (Neovim, bat, delta,
///   lsd) that look for COLORTERM to bypass the 256-color limit of the
///   screen-256color terminfo. ncurses ignores COLORTERM entirely — it keys
///   off the terminfo database for `$TERM` — so this is purely for those
///   direct-ANSI tools.
/// - Strips the invalid `LC_CTYPE=UTF-8` injected by macOS/OrbStack SSH hops.
///   "UTF-8" is a character encoding, not a valid POSIX locale name, which
///   causes `setlocale()` to fail and ncurses to crash or corrupt layout.
///   Removing it lets the child fall back to its own native locale default.
fn sanitize_child_environment(command: &mut CommandBuilder) {
    sanitize_child_environment_with_lc_ctype(command, std::env::var("LC_CTYPE").ok().as_deref());
}

/// Core of [`sanitize_child_environment`], parameterized over the incoming
/// `LC_CTYPE` value so the stripping logic is deterministically unit-testable
/// without mutating the process-global environment.
fn sanitize_child_environment_with_lc_ctype(command: &mut CommandBuilder, lc_ctype: Option<&str>) {
    command.env("TERM", CHILD_TERM);
    command.env("COLORTERM", CHILD_COLORTERM);
    if lc_ctype == Some("UTF-8") {
        command.env_remove("LC_CTYPE");
    }
}

impl Pty {
    pub fn spawn(command: CommandBuilder, size: PtySize) -> PtyResult<Self> {
        Self::spawn_with_scrollback(command, size, 0)
    }

    pub fn spawn_with_scrollback(
        mut command: CommandBuilder,
        size: PtySize,
        scrollback_len: usize,
    ) -> PtyResult<Self> {
        sanitize_child_environment(&mut command);
        let pty_system = native_pty_system();
        let pair = pty_system
            .openpty(size)
            .map_err(|err| wrap_err("openpty", err))?;
        let child = pair
            .slave
            .spawn_command(command)
            .map_err(|err| wrap_err("spawn_command", err))?;
        #[cfg(windows)]
        let job = {
            // Contain the child's process tree in a Job Object so kill paths
            // tear down grandchildren too (the Windows analogue of
            // `kill(-pgid, sig)`). Assignment happens immediately after spawn:
            // Win32 allows nested jobs, so membership in any parent job does
            // not block this. Once the child is assigned, every descendant it
            // subsequently spawns inherits job membership and dies with the
            // job (or on job-handle close).
            //
            // TODO(windows): assignment is post-spawn, so a descendant spawned
            // in the window between portable-pty's `CreateProcessW` returning
            // and our `AssignProcessToJobObject` call escapes the job and
            // would be orphaned by a kill. The formal zero-escape guarantee
            // requires spawning the child with `CREATE_SUSPENDED`, assigning
            // the suspended process to the job, then `ResumeThread`. portable-pty
            // 0.9.0 hardcodes the creation flags (`psuedocon.rs`:
            // `EXTENDED_STARTUPINFO_PRESENT | CREATE_UNICODE_ENVIRONMENT`) and
            // `CommandBuilder` exposes no creation-flags hook, so the suspended
            // spawn must be added upstream (or the crate forked). Until then
            // the guarantee is: descendants spawned after assignment are
            // contained; descendants spawned during the startup race are not.
            let job = JobObject::new().ok();
            if let (Some(job), Some(proc)) = (&job, child.as_raw_handle())
                && let Err(err) = unsafe { job.assign(proc) }
            {
                tracing::warn!(
                    "AssignProcessToJobObject failed: {err}; falling back to single-process kill"
                );
            }
            job
        };
        let reader = pair
            .master
            .try_clone_reader()
            .map_err(|err| wrap_err("try_clone_reader", err))?;
        let writer = PtyWriter::new(
            pair.master
                .take_writer()
                .map_err(|err| wrap_err("take_writer", err))?,
        );
        // The reader thread applies drain-synchronized resizes (reflow + ioctl),
        // so master must be shared with it.
        let master = Arc::new(Mutex::new(pair.master));
        let reader_master = Arc::clone(&master);
        let size_arc = Arc::new(Mutex::new(size));
        let reader_size = Arc::clone(&size_arc);
        let pending_resize = Arc::new(Mutex::new(None));
        let reader_pending_resize = Arc::clone(&pending_resize);
        #[cfg(unix)]
        let resize_wake = ResizeWake::new().map_err(|err| wrap_err("resize wake", err))?;
        let pending = Arc::new(Mutex::new(Vec::new()));
        let bytes_received = Arc::new(AtomicUsize::new(0));
        let last_bytes = Arc::new(Mutex::new(Vec::new()));
        let dsr_requested = Arc::new(AtomicBool::new(false));
        let reader_pending = Arc::clone(&pending);
        let reader_bytes = Arc::clone(&bytes_received);
        let reader_last = Arc::clone(&last_bytes);
        let reader_dsr = Arc::clone(&dsr_requested);
        let status_cb: StatusCallback = Arc::new(Mutex::new(None));
        let reader_status_cb = Arc::clone(&status_cb);
        let exited_emitted = Arc::new(AtomicBool::new(false));
        let reader_exited_emitted = Arc::clone(&exited_emitted);

        let pending_title = Arc::new(Mutex::new(None));
        let foreground_title = Arc::new(Mutex::new(None));
        let initial_parser = term_wm_vt100::Parser::new(size.rows, size.cols, scrollback_len);
        let tracker = std::sync::Arc::new(crate::PtyStateTracker::new(size.rows));
        let reader_tracker = std::sync::Arc::clone(&tracker);
        let shared_parser = Arc::new(Mutex::new(initial_parser));
        let dirty = Arc::new(AtomicBool::new(false));
        let dirty_cond = Arc::new((Mutex::new(()), Condvar::new()));
        let shutdown = Arc::new(AtomicBool::new(false));
        let reader_shutdown = Arc::clone(&shutdown);
        let reader_parser = Arc::clone(&shared_parser);
        let reader_dirty = Arc::clone(&dirty);
        let reader_dirty_cond = Arc::clone(&dirty_cond);
        let reader_pending_title = Arc::clone(&pending_title);
        #[cfg(unix)]
        let wake_read_fd = resize_wake.read_fd();
        let reader_handle = thread::spawn(move || {
            parser_read_loop(ParserReadLoopArgs {
                reader,
                pending: reader_pending,
                bytes_received: reader_bytes,
                last_bytes: reader_last,
                dsr_requested: reader_dsr,
                shared_parser: reader_parser,
                dirty: reader_dirty,
                dirty_cond: reader_dirty_cond,
                pending_title: reader_pending_title,
                status_cb: reader_status_cb,
                scrollback_len,
                osc52_text: None,
                clipboard: None,
                exited_emitted: reader_exited_emitted,
                tracker: reader_tracker,
                master: reader_master,
                size: reader_size,
                pending_resize: reader_pending_resize,
                #[cfg(unix)]
                wake_read_fd,
                shutdown: reader_shutdown,
            })
        });
        Ok(Self {
            #[cfg(unix)]
            master,
            writer,
            pending,
            bytes_received,
            last_bytes,
            dsr_requested,
            pending_title,
            foreground_title,
            last_fg_pid: 0,
            last_fg_check: Instant::now(),
            shared_parser,
            dirty,
            dirty_cond,
            size: size_arc,
            pending_resize,
            #[cfg(unix)]
            resize_wake,
            tracker,
            scrollback_len,
            child: Some(child),
            #[cfg(windows)]
            job,
            exited: false,
            exit_status: None,
            reader: Some(reader_handle),
            status_cb,
            exited_emitted,
            shutdown,
        })
    }

    /// Set a status callback invoked by the reader thread on data and exit.
    /// Uses `Arc<Mutex<>>` so the reader thread (which holds a clone) sees updates.
    ///
    /// If the child has already exited (reader thread detected EOF before the
    /// callback was registered, or ConPTY swallowed the EOF), fires the callback
    /// immediately so the async polling loop can process the exit state.
    pub fn set_status_callback(&mut self, cb: Option<Box<dyn Fn(PtyStatus) + Send + Sync>>) {
        let mut fire_cb = None;

        if let Ok(mut guard) = self.status_cb.lock() {
            if self.exited {
                self.exited_emitted.store(true, Ordering::Release);
                fire_cb = cb;
                *guard = None;
            } else if let Some(child) = self.child.as_mut()
                && let Ok(Some(status)) = child.try_wait()
            {
                self.exited = true;
                self.exit_status = Some(status);
                self.child = None;
                self.exited_emitted.store(true, Ordering::Release);
                fire_cb = cb;
                *guard = None;
            } else {
                *guard = cb;
            }
        }

        if let Some(reader) = &self.reader {
            reader.thread().unpark();
        }

        if let Some(cb_fn) = fire_cb {
            cb_fn(PtyStatus::Exited);
        }
    }

    /// Extract the child and reader handle for async reaping.
    /// After this call, the Pty is a shell — `update()` will no longer
    /// receive new data. Used by `Reaper::reap()`.
    pub fn into_parts(&mut self) -> PtyParts {
        self.shutdown.store(true, Ordering::Release);
        if let Some(reader) = &self.reader {
            reader.thread().unpark();
        }
        // Wake the reader so it notices `shutdown` and exits promptly (on
        // Windows this aborts the blocking ConPTY read; on Unix it nudges the
        // poll), so a later join on the returned handle does not hang.
        self.wake_reader();
        PtyParts {
            child: self.child.take(),
            reader_handle: self.reader.take(),
        }
    }

    /// Number of bytes received from the pty — always returns 0 after
    /// `into_parts()` has been called.
    pub fn reader_is_alive(&self) -> bool {
        self.reader.is_some()
    }

    /// Request a resize. The reader thread applies it (emulator reflow + OS
    /// `ioctl` / SIGWINCH) at the next pipe-drain boundary, so the grid width
    /// never changes while the shell is mid-draw and SIGWINCH is delivered only
    /// when the child is idle (drain-synchronized; no timers).
    pub fn resize(&mut self, size: PtySize) -> PtyResult<()> {
        // WORKAROUND: vt100 0.16.2 Grid::col_wrap (grid.rs:683) panics with a
        // subtraction overflow at cols=1; rows=1 causes similar issues. Clamp
        // the minimum so the PTY emulator doesn't crash when the terminal is
        // shrunk small.
        if size.rows < 2 || size.cols < 2 {
            return Ok(());
        }
        let mut pending = self
            .pending_resize
            .lock()
            .unwrap_or_else(|err| err.into_inner());
        if *pending == Some(size) {
            return Ok(()); // already requested
        }
        *pending = Some(size);
        drop(pending);
        self.wake_reader();
        Ok(())
    }

    /// Nudge the reader thread out of its blocking wait so it reaches the
    /// drain boundary and applies a pending resize promptly.
    ///
    /// Unix: writes the resize-wake self-pipe, which the reader polls alongside
    /// the PTY master fd.
    ///
    /// Windows: the ConPTY output pipe is a blocking anonymous pipe that cannot
    /// be polled or timed out (WSAPoll is sockets-only), so the reader parks in
    /// `ReadFile` while the pipe is idle and would never reach a drain boundary.
    /// `CancelSynchronousIo` aborts that in-flight read, which the reader
    /// recognises as a wake (`ERROR_OPERATION_ABORTED`), not an error. Safe to
    /// call when the reader is not currently blocked — the reader also re-checks
    /// `pending_resize` before its next blocking read.
    fn wake_reader(&mut self) {
        #[cfg(unix)]
        self.resize_wake.signal();
        #[cfg(windows)]
        if let Some(reader) = &self.reader {
            use std::os::windows::io::{AsHandle, AsRawHandle};
            // SAFETY: `handle` is the OS thread handle backing the reader's
            // `JoinHandle`, which is alive for as long as the `Pty` holds it.
            // It was created by `CreateThread` with `THREAD_ALL_ACCESS`, which
            // `CancelSynchronousIo` requires.
            unsafe {
                kernel32::CancelSynchronousIo(reader.as_handle().as_raw_handle());
            }
        }
    }

    pub fn write_bytes(&mut self, input: &[u8]) -> std::io::Result<()> {
        self.writer.write_bytes(input)
    }

    /// Cloneable handle for offloading blocking PTY writes to a dedicated
    /// thread (e.g. tokio's `spawn_blocking`). Never call the write from an
    /// async worker directly: a full kernel input buffer blocks the thread.
    pub fn writer_handle(&self) -> PtyWriter {
        self.writer.clone()
    }

    pub fn write_str(&mut self, input: &str) -> std::io::Result<()> {
        self.write_bytes(input.as_bytes())
    }

    pub fn take_pending_title(&self) -> Option<String> {
        let fg = self
            .foreground_title
            .lock()
            .unwrap_or_else(|err| err.into_inner())
            .clone();

        if fg.is_some() {
            // Process name is authoritative. Purge any stale OSC titles.
            let _ = self
                .pending_title
                .lock()
                .unwrap_or_else(|err| err.into_inner())
                .take();
            return fg;
        }

        self.pending_title
            .lock()
            .unwrap_or_else(|err| err.into_inner())
            .take()
    }

    fn poll_foreground(&mut self) {
        if self.last_fg_check.elapsed() >= FOREGROUND_POLL_INTERVAL {
            self.last_fg_check = Instant::now();
            if let Some(fg_pid) = self.foreground_pid()
                && fg_pid != self.last_fg_pid
            {
                self.last_fg_pid = fg_pid;
                let name = get_process_name(fg_pid);
                *self
                    .foreground_title
                    .lock()
                    .unwrap_or_else(|err| err.into_inner()) = name;
            }
        }
    }

    #[cfg(unix)]
    fn foreground_pid(&self) -> Option<u32> {
        self.master
            .lock()
            .unwrap_or_else(|err| err.into_inner())
            .process_group_leader()
            .map(|p| p as u32)
    }

    #[cfg(windows)]
    fn foreground_pid(&self) -> Option<u32> {
        let shell_pid = self.child.as_ref().and_then(|c| c.process_id())?;
        find_foreground_process_windows(shell_pid)
    }

    #[cfg(not(any(unix, windows)))]
    fn foreground_pid(&self) -> Option<u32> {
        None
    }

    /// Read pending bytes from the PTY reader thread (non-blocking).
    /// Used by the session server to forward raw bytes to remote clients.
    pub fn drain_pending(&mut self) -> Vec<u8> {
        let mut pending = self.pending.lock().unwrap_or_else(|err| err.into_inner());
        pending.split_off(0)
    }

    /// Drain all buffered output after the child has exited, first waiting
    /// (bounded by `grace`) for the reader thread to finish its EOF processing.
    ///
    /// The OS exit signal (`child.try_wait()`) can precede the reader thread's
    /// final read of the master fd, so draining immediately could truncate
    /// trailing bytes. The reader thread appends each chunk to `pending` before
    /// it reads EOF, so once the thread terminates `pending` is guaranteed
    /// complete. On Unix the reader EOFs within microseconds of process exit, so
    /// the wait is effectively free; on Windows ConPTY (where EOF can be
    /// swallowed and `has_exited()` relies on the `try_wait` fallback) the grace
    /// bounds the wait and we drain best-effort.
    pub fn drain_final_output(&mut self, grace: std::time::Duration) -> Vec<u8> {
        self.screen();
        if let Some(handle) = self.reader.as_ref() {
            let deadline = std::time::Instant::now() + grace;
            while !handle.is_finished() && std::time::Instant::now() < deadline {
                std::thread::sleep(std::time::Duration::from_millis(2));
            }
        }
        self.drain_pending()
    }

    pub fn screen_lines(&mut self) -> Vec<String> {
        self.screen(); // sync dirty state
        // Clone the screen out of the lock and drop the guard before any
        // string formatting, so the reader thread can keep ingesting bytes.
        let screen = self
            .shared_parser
            .lock()
            .unwrap_or_else(|err| err.into_inner())
            .screen()
            .clone();
        let contents = screen.contents();
        let mut lines: Vec<String> = contents.lines().map(|line| line.to_string()).collect();
        let rows = self.size.lock().unwrap_or_else(|err| err.into_inner()).rows as usize;
        if lines.len() < rows {
            lines.resize(rows, String::new());
        }
        lines
    }

    pub fn has_exited(&mut self) -> bool {
        if self.exited {
            return true;
        }
        let Some(child) = self.child.as_mut() else {
            return true;
        };
        match child.try_wait() {
            Ok(Some(status)) => {
                self.exited = true;
                self.exit_status = Some(status);
                self.child = None;

                // ConPTY pipes on Windows frequently swallow EOF, leaving the
                // reader thread blocked forever. Since this method is polled
                // every frame, we manually synthesize the exit callback here
                // when we detect the child process has died.
                // Use an atomic latch so the callback fires at most once —
                // the reader thread may also detect EOF and race to fire it.
                if let Ok(guard) = self.status_cb.lock()
                    && let Some(ref cb) = *guard
                    && !self.exited_emitted.swap(true, Ordering::AcqRel)
                {
                    cb(crate::PtyStatus::Exited);
                }

                true
            }
            Ok(None) => false,
            Err(_) => false,
        }
    }

    pub fn exit_status(&self) -> Option<portable_pty::ExitStatus> {
        self.exit_status.clone()
    }

    pub fn take_exit_status(&mut self) -> Option<portable_pty::ExitStatus> {
        self.exit_status.take()
    }

    /// Kill the child process if present.
    ///
    /// On Windows the whole contained tree is terminated via the Job Object
    /// (grandchildren included), falling back to single-process termination
    /// when containment was not established. The job handle is taken so its
    /// `Drop` (KILL_ON_JOB_CLOSE) is the last line of defense for stragglers.
    pub fn kill_child(&mut self) -> PtyResult<()> {
        #[cfg(windows)]
        if let Some(job) = self.job.take() {
            let _ = job.terminate(1);
        }
        if let Some(mut child) = self.child.take() {
            child.kill().map_err(|err| wrap_err("kill", err))?;
            self.exited = true;
            self.child = None;
        }
        Ok(())
    }

    /// Process group leader (pgid) of the PTY child's session, when known.
    ///
    /// On Unix the child is started via `setsid()` (portable-pty), so its pgid
    /// equals its pid; a negative-pgid `kill(2)` then signals the whole
    /// process tree. Returns `None` where the platform does not expose it
    /// (e.g. Windows ConPTY).
    #[cfg(unix)]
    pub fn process_group_id(&self) -> Option<i32> {
        self.master
            .lock()
            .unwrap_or_else(|err| err.into_inner())
            .process_group_leader()
    }

    /// Send a signal to the child's entire process group, not just the leader.
    ///
    /// Strictly non-blocking: translates to a single `kill(-pgid, signal)` and
    /// returns immediately. This is mechanism only — the caller (the daemon
    /// supervisor) owns any escalation policy (grace timers, SIGTERM→SIGKILL
    /// sequencing, exited-state arbitration). No temporal waits live here.
    ///
    /// Returns an error when no process group is known on this platform; the
    /// supervisor falls back to `kill_child()` (single process).
    #[cfg(unix)]
    pub fn signal_process_group(&self, signal: i32) -> PtyResult<()> {
        let Some(pgid) = self
            .master
            .lock()
            .unwrap_or_else(|err| err.into_inner())
            .process_group_leader()
        else {
            return Err(wrap_err(
                "signal_process_group",
                std::io::Error::new(
                    std::io::ErrorKind::NotFound,
                    "PTY child has no process group",
                ),
            ));
        };
        let ret = unsafe { libc::kill(-pgid, signal) };
        if ret == -1 {
            return Err(wrap_err(
                "signal_process_group",
                std::io::Error::last_os_error(),
            ));
        }
        Ok(())
    }

    pub fn size(&self) -> PtySize {
        *self.size.lock().unwrap_or_else(|err| err.into_inner())
    }

    /// Sync dirty state and handle DSR/foreground polling.
    /// The caller should then lock `shared_parser()` directly for cell access.
    pub fn screen(&mut self) {
        self.poll_foreground();
        if self.dirty.swap(false, Ordering::Acquire) {
            // Send DSR response if requested by the reader thread.
            if self.dsr_requested.swap(false, Ordering::Relaxed) {
                let parser = self.shared_parser.lock().unwrap();
                let (row, col) = parser.screen().cursor_position();
                drop(parser);
                let response = format!("\x1b[{};{}R", row.saturating_add(1), col.saturating_add(1));
                let _ = self.write_bytes(response.as_bytes());
            }
            // Acquire the lock to prevent lost wakeups on the condition variable
            let (lock, cvar) = &*self.dirty_cond;
            let _guard = lock.lock().unwrap();
            cvar.notify_all();
        }
    }

    /// Sync dirty state and return the full terminal snapshot.
    /// Combines screen() (which clears dirty and wakes the reader)
    /// with a formatted snapshot of the current parser state.
    pub fn generate_snapshot(&mut self) -> Vec<u8> {
        self.screen();
        // Clone the screen out of the lock and drop the guard before the
        // O(rows x cols) escape-code formatting, so the reader thread can keep
        // ingesting bytes while the snapshot is built.
        self.shared_parser
            .lock()
            .unwrap_or_else(|err| err.into_inner())
            .screen()
            .clone()
            .state_formatted()
    }

    pub fn bytes_received(&self) -> usize {
        self.bytes_received.load(Ordering::Relaxed)
    }

    pub fn last_bytes_text(&self) -> String {
        let bytes = self
            .last_bytes
            .lock()
            .map(|buf| buf.clone())
            .unwrap_or_default();
        bytes_to_debug_text(&bytes, 32)
    }

    pub fn scrollback(&mut self) -> usize {
        self.screen(); // sync dirty state
        let parser = self.shared_parser.lock().unwrap();
        parser.screen().scrollback()
    }

    pub fn set_scrollback(&mut self, rows: usize) {
        let max = self.scrollback_len;
        let mut parser = self.shared_parser.lock().unwrap();
        parser.screen_mut().set_scrollback(rows.min(max));
    }

    pub fn scrollback_len(&self) -> usize {
        self.scrollback_len
    }

    pub fn max_scrollback(&mut self) -> usize {
        let max_sb = self.scrollback_len;
        if max_sb == 0 {
            return 0;
        }
        let mut parser = self.shared_parser.lock().unwrap();
        let screen = parser.screen_mut();
        let current = screen.scrollback();
        screen.set_scrollback(max_sb);
        let max = screen.scrollback();
        screen.set_scrollback(current);
        max
    }

    pub fn alternate_screen(&mut self) -> bool {
        self.screen(); // sync dirty state
        let parser = self.shared_parser.lock().unwrap();
        parser.screen().alternate_screen()
    }

    /// Return an `Arc<dyn DirectInputTracker>` for this PTY's state tracker.
    /// Used by the window manager to auto-enable Direct Mode when the
    /// application enters alternate screen, enables mouse tracking, etc.
    pub fn direct_input_tracker(&self) -> std::sync::Arc<dyn crate::DirectInputTracker> {
        self.tracker.clone()
    }
}

impl Drop for Pty {
    fn drop(&mut self) {
        self.shutdown.store(true, Ordering::Release);
        if let Some(reader) = &self.reader {
            reader.thread().unpark();
        }
        self.wake_reader();
    }
}

/// Configuration and shared state for the PTY reader thread.
struct ParserReadLoopArgs {
    reader: Box<dyn Read + Send>,
    pending: Arc<Mutex<Vec<u8>>>,
    bytes_received: Arc<AtomicUsize>,
    last_bytes: Arc<Mutex<Vec<u8>>>,
    dsr_requested: Arc<AtomicBool>,
    shared_parser: Arc<Mutex<term_wm_vt100::Parser>>,
    dirty: Arc<AtomicBool>,
    dirty_cond: Arc<(std::sync::Mutex<()>, Condvar)>,
    pending_title: Arc<Mutex<Option<String>>>,
    status_cb: StatusCallback,
    scrollback_len: usize,
    /// Shared latch: taken once by whichever path detects exit first.
    exited_emitted: Arc<AtomicBool>,
    /// Test-only hook: when `Some`, the extracted OSC 52 text is written here
    /// in addition to the real clipboard, so tests can assert the value.
    osc52_text: Option<Arc<Mutex<Option<String>>>>,
    /// Clipboard to relay extracted OSC 52 sequences through. `None` in
    /// production (the loop constructs a default handle); tests inject an
    /// isolated, headless [`Clipboard`] so the relay can be exercised without
    /// touching the real system clipboard or process-global default buffer.
    clipboard: Option<Clipboard>,
    /// Application state tracker — shared via Arc with Pty.
    tracker: std::sync::Arc<crate::PtyStateTracker>,
    /// Shared master for drain-synchronized `ioctl` resizes applied by the reader.
    master: Arc<Mutex<Box<dyn MasterPty + Send>>>,
    /// Current emulator size (shared; updated by the reader when it applies a resize).
    size: Arc<Mutex<PtySize>>,
    /// Resize requested by the UI, applied at the next pipe-drain boundary.
    pending_resize: Arc<Mutex<Option<PtySize>>>,
    /// Read end of the resize wake self-pipe (polled alongside the PTY fd).
    #[cfg(unix)]
    wake_read_fd: RawFd,
    /// Set by `into_parts`/`Drop`: the reader exits its loop ASAP. On Windows
    /// in particular the blocking ConPTY read is otherwise uninterruptible.
    shutdown: Arc<AtomicBool>,
}

fn parser_read_loop(args: ParserReadLoopArgs) {
    let ParserReadLoopArgs {
        mut reader,
        pending,
        bytes_received,
        last_bytes,
        dsr_requested,
        shared_parser,
        dirty,
        dirty_cond,
        pending_title,
        status_cb,
        scrollback_len: _scrollback_len,
        osc52_text,
        clipboard,
        exited_emitted,
        tracker,
        master,
        size,
        pending_resize,
        #[cfg(unix)]
        wake_read_fd,
        shutdown,
    } = args;
    let mut prev_tail: [u8; HISTORY_TAIL_LEN] = [0; HISTORY_TAIL_LEN];
    let mut buf = [0u8; PTY_READ_BUF_SIZE];
    let mut osc52 = Osc52Extractor::new();
    let mut bytes_since_render = 0usize;
    let mut vte_parser = vte::Parser::new();
    let tracker_for_adapter = std::sync::Arc::clone(&tracker);
    let mut tracker_adapter = PtyPerformAdapter::new(tracker_for_adapter);
    // One clipboard handle for the whole reader lifetime: re-initialising
    // arboard per OSC 52 sequence would repeat the display-server handshake
    // on every copy event.  Each extracted sequence is relayed synchronously
    // (no debounce) so the tail payload of a burst is never dropped.
    //
    // Initialised lazily on the first OSC 52 sequence: arboard's handshake can
    // block for 500ms+ (macOS pasteboard / Wayland), and we must not stall the
    // reader loop's startup (and thus the first drain-synchronized resize) on it.
    let mut clipboard: Option<Clipboard> = clipboard;
    const IO_BURST_BUDGET: usize = 256 * 1024; // 256 KB
    // The pty master fd never changes — fetch it once to avoid a Mutex lock on
    // every drain iteration (the UI thread and apply_resize also lock `master`).
    #[cfg(unix)]
    let master_fd = master
        .lock()
        .unwrap_or_else(|err| err.into_inner())
        .as_raw_fd()
        .unwrap_or(-1);
    'reader: loop {
        if shutdown.load(Ordering::Acquire) {
            break 'reader;
        }
        // Block until the PTY master or the resize-wake is readable, so a
        // resize request is noticed even while the pipe is idle.
        #[cfg(unix)]
        {
            let mut pollfds = [
                libc::pollfd {
                    fd: master_fd,
                    events: libc::POLLIN,
                    revents: 0,
                },
                libc::pollfd {
                    fd: wake_read_fd,
                    events: libc::POLLIN,
                    revents: 0,
                },
            ];
            let _ = unsafe { libc::poll(pollfds.as_mut_ptr(), pollfds.len() as libc::nfds_t, -1) };
            if pollfds[1].revents & libc::POLLIN != 0 {
                clear_wake(wake_read_fd);
            }
        }

        // Drain: keep reading until the pipe is empty (or the starvation bound),
        // then apply any pending resize at the drain boundary.
        #[cfg(unix)]
        let mut drain_bytes = 0usize;
        // Clippy: on Windows every path through this loop breaks — the blocking
        // ConPTY read cannot drain multiple chunks per iteration, so the shape
        // is deliberately a single read + drain boundary (the boundary code is
        // shared with Unix below).
        #[allow(clippy::never_loop)]
        loop {
            // (Windows) The ConPTY read is a blocking anonymous pipe that cannot
            // be polled or timed out. A `CancelSynchronousIo` wake can be lost
            // if it lands while the reader is between reads, so check for a
            // pending resize (and shutdown) before blocking — the boundary apply
            // below then reflows the grid without waiting for the next chunk of
            // output.
            #[cfg(windows)]
            {
                let has_pending = pending_resize
                    .lock()
                    .unwrap_or_else(|err| err.into_inner())
                    .is_some();
                if has_pending || shutdown.load(Ordering::Acquire) {
                    break;
                }
            }
            // (Unix) zero-timeout availability check: pipe empty → drain boundary.
            #[cfg(unix)]
            {
                let mut pfd = libc::pollfd {
                    fd: master_fd,
                    events: libc::POLLIN,
                    revents: 0,
                };
                let pr = unsafe { libc::poll(&mut pfd, 1, 0) };
                let hup = pfd.revents & (libc::POLLHUP | libc::POLLERR) != 0;
                if pr <= 0 || (pfd.revents & libc::POLLIN == 0 && !hup) {
                    break;
                }
            }
            match reader.read(&mut buf) {
                Ok(0) => {
                    // EOF — child exited. Flush any buffered OSC 52 payload
                    // (Windows ConPTY may have consumed the terminator).
                    if let Some(text) = osc52.finish() {
                        if clipboard.is_none() {
                            clipboard = Some(Clipboard::new());
                        }
                        if let Some(clip) = clipboard.as_mut() {
                            clip.set(&text);
                        }
                        if let Some(ref capture) = osc52_text {
                            *capture.lock().unwrap_or_else(|err| err.into_inner()) = Some(text);
                        }
                    }
                    // Send wakeup for final screen, then exited.
                    let guard = status_cb.lock().unwrap_or_else(|err| err.into_inner());
                    if let Some(ref cb) = *guard {
                        cb(crate::PtyStatus::Wakeup);
                        if !exited_emitted.swap(true, Ordering::AcqRel) {
                            cb(crate::PtyStatus::Exited);
                        }
                    }
                    break 'reader;
                }
                Ok(n) => {
                    bytes_received.fetch_add(n, Ordering::Relaxed);
                    bytes_since_render += n;
                    // DSR detection: 65536-byte reads make cross-chunk splitting
                    // of the 4-byte \x1b[6n pattern vanishingly unlikely.
                    if buf[..n].windows(DSR_PATTERN_LEN).any(|w| w == b"\x1b[6n") {
                        dsr_requested.store(true, Ordering::Relaxed);
                    }
                    let mut last = last_bytes.lock().unwrap_or_else(|err| err.into_inner());
                    last.clear();
                    last.extend_from_slice(&buf[..n]);
                    let mut p = pending.lock().unwrap_or_else(|err| err.into_inner());
                    p.extend_from_slice(&buf[..n]);
                    // Cap pending to prevent unbounded growth when no
                    // consumer calls drain_pending() (local terminal mode).
                    const PENDING_CAP: usize = 1024 * 1024; // 1 MB
                    if p.len() > PENDING_CAP {
                        p.clear();
                    }

                    // Process bytes through the application state tracker
                    // (alternate screen, mouse tracking, margins — atomics, no lock).
                    let prev_mode = tracker.direct_input_mode();
                    vte_parser.advance(&mut tracker_adapter, &buf[..n]);
                    let new_mode = tracker.direct_input_mode();
                    if prev_mode != new_mode {
                        tracing::info!(
                            "[STAGE 1] PTY routing flipped: {:?} -> {:?}",
                            prev_mode,
                            new_mode
                        );
                        let guard = status_cb.lock().unwrap_or_else(|err| err.into_inner());
                        if let Some(ref cb) = *guard {
                            cb(crate::PtyStatus::DirectInputChanged(new_mode));
                        } else {
                            tracing::error!(
                                "[STAGE 1] status_cb is NONE when transition occurred!"
                            );
                        }
                    }

                    // Process bytes directly into the shared parser.
                    {
                        let mut shared =
                            shared_parser.lock().unwrap_or_else(|err| err.into_inner());
                        shared.process(&buf[..n]);
                    }
                    esc_trace_chunk(&buf[..n]);

                    if let Some(title) = extract_osc_title(&buf[..n]) {
                        let mut guard = pending_title.lock().unwrap_or_else(|err| err.into_inner());
                        *guard = Some(title);
                    }
                    // Intercept OSC 52 clipboard sequences (cross-chunk buffering).
                    // Relay each extracted sequence synchronously via the hoisted
                    // handle — no debounce, so the tail payload is never dropped.
                    // The handle is lazy-initialised on the first sequence so the
                    // reader loop's startup is never blocked on arboard's handshake.
                    //
                    // TODO: OSC 52 interception currently runs unconditionally —
                    // including in Direct Input Mode, where mouse-managed clipboard
                    // handling is otherwise delegated to the running app. Decide
                    // whether (and how) to gate this relay (also considering the
                    // nested term-session client relay) and file a GitHub issue to
                    // track it.
                    if let Some(text) = osc52.push(&buf[..n], &prev_tail) {
                        if clipboard.is_none() {
                            clipboard = Some(Clipboard::new());
                        }
                        if let Some(clip) = clipboard.as_mut() {
                            clip.set(&text);
                        }
                        if let Some(ref capture) = osc52_text {
                            *capture.lock().unwrap_or_else(|err| err.into_inner()) = Some(text);
                        }
                    }

                    // Update prev_tail for next iteration's cross-chunk detection.
                    if n >= HISTORY_TAIL_LEN {
                        prev_tail.copy_from_slice(&buf[n - HISTORY_TAIL_LEN..n]);
                    } else if n > 0 {
                        prev_tail.rotate_left(n);
                        prev_tail[HISTORY_TAIL_LEN - n..].copy_from_slice(&buf[..n]);
                    }

                    // Edge-triggered wakeup: only notify on false→true transition.
                    // Prevents flooding the IPC channel with thousands of redundant
                    // PtyWakeup messages per second at unthrottled ingestion speeds.
                    if !dirty.swap(true, Ordering::AcqRel) {
                        let guard = status_cb.lock().unwrap_or_else(|err| err.into_inner());
                        if let Some(ref cb) = *guard {
                            cb(crate::PtyStatus::Wakeup);
                        }
                        // Reset budget because a new render cycle has begun
                        bytes_since_render = 0;
                    }

                    // I/O burst budget: when the reader has ingested more than
                    // IO_BURST_BUDGET bytes without a render, wait on the Condvar
                    // until the UI thread clears dirty. This prevents a single
                    // reader thread from consuming 100% CPU on infinite streams.
                    if bytes_since_render >= IO_BURST_BUDGET {
                        let (lock, cvar) = &*dirty_cond;
                        let mut guard = lock.lock().unwrap_or_else(|err| err.into_inner());
                        while dirty.load(Ordering::Acquire) {
                            guard = cvar.wait(guard).unwrap_or_else(|err| err.into_inner());
                        }
                        bytes_since_render = 0;
                    }

                    // Drain-sync: break at the starvation bound so a continuous
                    // stream (which never returns WouldBlock) still applies the
                    // pending resize and keeps the grid attached to the window.
                    #[cfg(unix)]
                    {
                        drain_bytes += n;
                        if drain_bytes >= MAX_DRAIN_BYTES {
                            break;
                        }
                    }

                    // Loop back to read() — no parking, no cloning, no render_ready check.
                    // Lock contention is expected under load: the reader will block
                    // on the mutex while the main thread holds it during render.
                    // This is intentional mechanical backpressure.
                }
                Err(err) => {
                    // Windows: `wake_reader` aborts the blocking ConPTY read via
                    // `CancelSynchronousIo`; the aborted `ReadFile` returns
                    // `ERROR_OPERATION_ABORTED` (995). That is a resize wake, not a
                    // fatal error — break to the drain boundary so the pending
                    // resize is applied. (On Unix `raw_os_error` can never be 995,
                    // so this branch is inert there.)
                    if err.raw_os_error() == Some(ERROR_OPERATION_ABORTED) {
                        break;
                    }
                    let guard = status_cb.lock().unwrap_or_else(|err| err.into_inner());
                    if let Some(ref cb) = *guard
                        && !exited_emitted.swap(true, Ordering::AcqRel)
                    {
                        cb(crate::PtyStatus::Exited);
                    }
                    break 'reader;
                }
            }
            // (Windows) one blocking read per outer iteration (no non-blocking drain
            // until the ConPTY handle is pollable — resizes apply at this boundary).
            #[cfg(windows)]
            {
                break;
            }
        }
        // DRAIN BOUNDARY — pipe empty (or starvation bound): apply any pending
        // resize here, so the grid is reflowed and SIGWINCH is delivered only
        // when the shell is not mid-write.
        let new_size = pending_resize
            .lock()
            .unwrap_or_else(|err| err.into_inner())
            .take();
        if let Some(new_size) = new_size {
            apply_resize(new_size, &shared_parser, &tracker, &master, &size);
        }
    }
}

/// Apply a drain-synchronized resize on the reader thread: reflow the vt100
/// grid (with the shrink `ESC[S` cursor push), update the tracker and the shared
/// size, then issue the OS resize (`ioctl` / SIGWINCH) — all so the grid width
/// never changes mid-shell-write.
fn apply_resize(
    new_size: PtySize,
    shared_parser: &Arc<Mutex<term_wm_vt100::Parser>>,
    tracker: &Arc<crate::PtyStateTracker>,
    master: &Arc<Mutex<Box<dyn MasterPty + Send>>>,
    size: &Arc<Mutex<PtySize>>,
) {
    let old_rows = {
        let parser = shared_parser.lock().unwrap_or_else(|err| err.into_inner());
        parser.screen().size().0
    };
    let mut guard = shared_parser.lock().unwrap_or_else(|err| err.into_inner());
    if new_size.rows < old_rows && !tracker.has_custom_margins() {
        let (cursor_row, _) = guard.screen().cursor_position();
        if cursor_row >= new_size.rows {
            let scroll_lines = cursor_row - new_size.rows + 1;
            let seq = format!("\x1b[{scroll_lines}S");
            guard.process(seq.as_bytes());
        }
    }
    guard.screen_mut().set_size(new_size.rows, new_size.cols);
    drop(guard);
    tracker.resize(new_size.rows);
    *size.lock().unwrap_or_else(|err| err.into_inner()) = new_size;
    if let Err(e) = master
        .lock()
        .unwrap_or_else(|err| err.into_inner())
        .resize(new_size)
    {
        tracing::warn!("drain-sync PTY resize failed: {e}");
    }
}

fn wrap_err<E: std::fmt::Display>(
    stage: &'static str,
    err: E,
) -> Box<dyn std::error::Error + Send + Sync> {
    Box::new(std::io::Error::other(format!("pty {stage} failed: {err}")))
}

fn bytes_to_debug_text(bytes: &[u8], max_len: usize) -> String {
    let mut out = String::new();
    for &b in bytes.iter().take(max_len) {
        match b {
            b'\r' => out.push_str("\\r"),
            b'\n' => out.push_str("\\n"),
            b'\t' => out.push_str("\\t"),
            0x20..=0x7e => out.push(b as char),
            _ => out.push_str(&format!("\\x{:02x}", b)),
        }
    }
    out
}

/// Get the process name for a given PID. On macOS uses `proc_name` from
/// libproc. On Linux reads `/proc/<pid>/comm`. On other platforms returns None.
#[cfg(target_os = "macos")]
fn get_process_name(pid: u32) -> Option<String> {
    let mut name = [0u8; PROC_NAME_BUF_SIZE];
    let result = unsafe {
        libc::proc_name(
            pid as libc::c_int,
            name.as_mut_ptr() as *mut libc::c_void,
            name.len() as u32,
        )
    };
    if result > 0 {
        let len = name.iter().position(|&b| b == 0).unwrap_or(name.len());
        Some(String::from_utf8_lossy(&name[..len]).into_owned())
    } else {
        None
    }
}

#[cfg(target_os = "linux")]
fn get_process_name(pid: u32) -> Option<String> {
    let path = format!("/proc/{pid}/comm");
    std::fs::read_to_string(&path)
        .ok()
        .map(|s| s.trim().to_string())
}

#[cfg(windows)]
fn get_process_name(pid: u32) -> Option<String> {
    use std::ffi::OsString;
    use std::os::windows::ffi::OsStringExt;

    const PROCESS_QUERY_LIMITED_INFORMATION: u32 = 0x1000;
    let handle = unsafe { kernel32::OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION, 0, pid) };
    if handle == 0 {
        return None;
    }

    let mut buf = [0u16; 260];
    let mut size = buf.len() as u32;
    let result =
        unsafe { kernel32::QueryFullProcessImageNameW(handle, 0, buf.as_mut_ptr(), &mut size) };
    unsafe {
        kernel32::CloseHandle(handle);
    }

    if result == 0 {
        return None;
    }
    let path = OsString::from_wide(&buf[..size as usize]);
    std::path::Path::new(&path)
        .file_stem()
        .map(|s| s.to_string_lossy().into_owned())
}

#[cfg(windows)]
fn find_foreground_process_windows(shell_pid: u32) -> Option<u32> {
    let snapshot = unsafe { kernel32::CreateToolhelp32Snapshot(0x00000002, 0) };
    if snapshot == kernel32::INVALID_HANDLE_VALUE {
        return None;
    }

    let mut children: Vec<(u32, u32)> = Vec::new();
    let mut entry = std::mem::MaybeUninit::<kernel32::PROCESSENTRY32W>::zeroed();

    unsafe {
        (*entry.as_mut_ptr()).dwSize = std::mem::size_of::<kernel32::PROCESSENTRY32W>() as u32;
        if kernel32::Process32FirstW(snapshot, entry.as_mut_ptr()) != 0 {
            loop {
                let e = entry.assume_init();
                children.push((e.th32ProcessID, e.th32ParentProcessID));
                if kernel32::Process32NextW(snapshot, entry.as_mut_ptr()) == 0 {
                    break;
                }
            }
        }
        kernel32::CloseHandle(snapshot);
    }

    let mut current = shell_pid;
    loop {
        let next = children
            .iter()
            .find(|&&(pid, parent)| parent == current && pid != current)
            .map(|&(pid, _)| pid);
        match next {
            Some(next) => current = next,
            None => break,
        }
    }

    Some(current)
}

#[cfg(windows)]
mod kernel32 {
    pub const INVALID_HANDLE_VALUE: isize = -1;

    #[repr(C)]
    #[derive(Copy, Clone)]
    #[allow(non_snake_case)]
    pub struct PROCESSENTRY32W {
        pub dwSize: u32,
        pub cntUsage: u32,
        pub th32ProcessID: u32,
        pub th32DefaultHeapID: usize,
        pub th32ModuleID: u32,
        pub cntThreads: u32,
        pub th32ParentProcessID: u32,
        pub pcPriClassBase: i32,
        pub dwFlags: u32,
        pub szExeFile: [u16; 260],
    }

    #[allow(non_snake_case)]
    unsafe extern "system" {
        pub fn CreateToolhelp32Snapshot(dwFlags: u32, th32ProcessID: u32) -> isize;
        pub fn Process32FirstW(hSnapshot: isize, lppe: *mut PROCESSENTRY32W) -> i32;
        pub fn Process32NextW(hSnapshot: isize, lppe: *mut PROCESSENTRY32W) -> i32;
        pub fn CloseHandle(hObject: isize) -> i32;
        pub fn OpenProcess(dwDesiredAccess: u32, bInheritHandle: i32, dwProcessId: u32) -> isize;
        pub fn QueryFullProcessImageNameW(
            hProcess: isize,
            dwFlags: u32,
            lpExeName: *mut u16,
            lpdwSize: *mut u32,
        ) -> i32;
        /// Abort pending synchronous I/O issued by a thread — used to wake the
        /// reader out of a blocking ConPTY read on a resize request.
        pub fn CancelSynchronousIo(hThread: *mut std::ffi::c_void) -> i32;
    }
}

#[cfg(not(any(target_os = "macos", target_os = "linux", windows)))]
fn get_process_name(_pid: u32) -> Option<String> {
    None
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::pty::StatusCallback;
    use std::io;
    #[cfg(windows)]
    use std::io::Cursor;
    use std::io::Write;
    use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
    use std::sync::{Arc, Mutex, RwLock};

    /// Returns a platform-appropriate dummy executable for PTY plumbing tests.
    /// On Unix, `cat` blocks on stdin and echoes output. On Windows, `cmd.exe`
    /// blocks on stdin and keeps the ConPTY alive.
    fn get_test_executable() -> &'static str {
        #[cfg(target_os = "windows")]
        {
            "cmd.exe"
        }
        #[cfg(not(target_os = "windows"))]
        {
            "cat"
        }
    }

    /// Locate the `term-session-mock` binary so the Job Object test can spawn
    /// a child that itself forks a grandchild. Resolution is delegated to the
    /// shared helper in the mock crate's library (see
    /// `term_session_mock::get_mock_bin`), which is a dev-dependency of this
    /// crate so the binary is always built for tests.
    #[cfg(windows)]
    fn get_mock_bin() -> std::path::PathBuf {
        term_session_mock::get_mock_bin()
    }

    // ── bytes_to_debug_text ──────────────────────────────────────────

    #[test]
    fn bytes_to_debug_text_empty() {
        assert_eq!(bytes_to_debug_text(b"", 32), "");
    }

    #[test]
    fn bytes_to_debug_text_printable_passthrough() {
        assert_eq!(bytes_to_debug_text(b"hello world", 32), "hello world");
    }

    #[test]
    fn bytes_to_debug_text_encodes_control_and_nonprint() {
        let data = b"a\nb\tc\r\x01\xff";
        let s = bytes_to_debug_text(data, 32);
        assert!(s.contains("a\\nb\\tc\\r"));
        assert!(s.contains("\\x01"));
        assert!(s.contains("\\xff"));
    }

    #[test]
    fn bytes_to_debug_text_truncates_at_max_len() {
        let long = b"abcdefghijklmnopqrstuvwxyz";
        assert_eq!(bytes_to_debug_text(long, 5).len(), 5);
    }

    #[test]
    fn bytes_to_debug_text_short_max_len() {
        let s = bytes_to_debug_text(b"hello", 0);
        assert_eq!(s, "");
    }

    #[test]
    fn bytes_to_debug_text_all_control_chars() {
        let data: Vec<u8> = (0..32).collect();
        let s = bytes_to_debug_text(&data, 64);
        // Characters 0x00-0x08, 0x0b-0x1f use \xNN; 0x09=\t, 0x0a=\n, 0x0d=\r
        for i in 0..32u8 {
            let expected = match i {
                0x09 => 't',
                0x0a => 'n',
                0x0d => 'r',
                _ => continue,
            };
            assert!(
                s.contains(&format!("\\{}", expected)),
                "missing named escape for 0x{i:02x}"
            );
        }
        // Verify a few non-special controls use \xNN format
        assert!(s.contains("\\x00"));
        assert!(s.contains("\\x01"));
        assert!(s.contains("\\x1b"));
        assert!(s.contains("\\x1f"));
    }

    // ── parser_read_loop ─────────────────────────────────────────────
    fn make_parser_test_args(payload: &[u8]) -> ParserReadLoopArgs {
        // A real PTY is required so the drain-synchronized reader loop can poll
        // the master fd. On Unix the payload is produced by a `printf` child
        // written to the slave (child output direction → master reader), which
        // then exits to EOF the loop. On Windows the loop is the non-polling
        // fallback, so a synchronous Cursor reader works.
        let pty_system = native_pty_system();
        let pair = pty_system
            .openpty(PtySize {
                rows: 24,
                cols: 80,
                pixel_width: 0,
                pixel_height: 0,
            })
            .expect("openpty");
        #[cfg(unix)]
        let reader = {
            // Escape the payload for `sh -c 'printf …'` so it is written to the
            // slave verbatim; the child exits immediately afterward.
            let mut escaped = String::from("printf '");
            for &b in payload {
                match b {
                    b'\'' => escaped.push_str("'\\''"),
                    0x1b => escaped.push_str("\\033"),
                    b'\n' => escaped.push_str("\\n"),
                    b'\r' => escaped.push_str("\\r"),
                    b'\t' => escaped.push_str("\\t"),
                    0x20..=0x7e => escaped.push(b as char),
                    _ => escaped.push_str(&format!("\\{:03o}", b)),
                }
            }
            escaped.push('\'');
            let mut builder = CommandBuilder::new("sh");
            builder.arg("-c");
            builder.arg(escaped);
            let _child = pair
                .slave
                .spawn_command(builder)
                .expect("spawn printf child");
            pair.master.try_clone_reader().expect("reader")
        };
        #[cfg(windows)]
        let reader = Box::new(Cursor::new(payload.to_vec()));
        let master = Arc::new(Mutex::new(pair.master));
        drop(pair.slave);
        #[cfg(unix)]
        let wake_read_fd = {
            // Keep the wake pipe alive for the reader's poll set; dropping it
            // here would close the fd and make `poll` return POLLNVAL (busy-spin).
            let wake = Box::leak(Box::new(ResizeWake::new().expect("resize wake")));
            wake.read_fd()
        };
        ParserReadLoopArgs {
            reader,
            pending: Arc::new(Mutex::new(Vec::new())),
            bytes_received: Arc::new(AtomicUsize::new(0)),
            last_bytes: Arc::new(Mutex::new(Vec::new())),
            dsr_requested: Arc::new(AtomicBool::new(false)),
            shared_parser: Arc::new(Mutex::new(term_wm_vt100::Parser::new(24, 80, 0))),
            dirty: Arc::new(AtomicBool::new(false)),
            dirty_cond: Arc::new((Mutex::new(()), Condvar::new())),
            pending_title: Arc::new(Mutex::new(None)),
            status_cb: Arc::new(Mutex::new(None)),
            scrollback_len: 0,
            exited_emitted: Arc::new(AtomicBool::new(false)),
            osc52_text: None,
            clipboard: None,
            tracker: std::sync::Arc::new(crate::PtyStateTracker::new(24)),
            master,
            size: Arc::new(Mutex::new(PtySize {
                rows: 24,
                cols: 80,
                pixel_width: 0,
                pixel_height: 0,
            })),
            pending_resize: Arc::new(Mutex::new(None)),
            #[cfg(unix)]
            wake_read_fd,
            shutdown: Arc::new(AtomicBool::new(false)),
        }
    }

    #[test]
    fn parser_read_loop_reads_and_sets_pending_and_last() {
        let payload = b"hello\r\n\x1b[6nworld";
        let args = make_parser_test_args(payload);
        let pending = Arc::clone(&args.pending);
        let bytes_received = Arc::clone(&args.bytes_received);
        let last_bytes = Arc::clone(&args.last_bytes);
        let dsr_requested = Arc::clone(&args.dsr_requested);
        let dirty = Arc::clone(&args.dirty);

        parser_read_loop(args);

        let p = pending.lock().unwrap();
        assert!(!p.is_empty());
        assert!(bytes_received.load(Ordering::Relaxed) > 0);
        let last = last_bytes.lock().unwrap();
        assert!(!last.is_empty());
        assert!(dsr_requested.load(Ordering::Relaxed));
        assert!(dirty.load(Ordering::Relaxed));
    }

    #[test]
    fn parser_read_loop_empty_input() {
        let args = make_parser_test_args(b"");
        let pending = Arc::clone(&args.pending);
        let bytes_received = Arc::clone(&args.bytes_received);
        let last_bytes = Arc::clone(&args.last_bytes);
        let dsr_requested = Arc::clone(&args.dsr_requested);
        let dirty = Arc::clone(&args.dirty);

        parser_read_loop(args);

        let p = pending.lock().unwrap();
        assert!(p.is_empty());
        assert_eq!(bytes_received.load(Ordering::Relaxed), 0);
        let last = last_bytes.lock().unwrap();
        assert!(last.is_empty());
        assert!(!dsr_requested.load(Ordering::Relaxed));
        assert!(!dirty.load(Ordering::Relaxed));
    }

    /// Regression: the emulator term-wm builds against must honor DECAWM
    /// (`ESC[?7l`, autowrap off). Feeding `\x1b[?7l` + 85 chars through the
    /// production ingestion loop (`parser_read_loop`) must clamp the cursor to
    /// the right margin instead of wrapping onto the next row. Guards the
    /// workspace-level integration point: without DECAWM support the emulator
    /// would ignore the toggle and wrap the 81st character to row 1, failing
    /// these assertions.
    #[test]
    fn parser_read_loop_decawn_off_does_not_wrap() {
        let mut payload = Vec::new();
        payload.extend_from_slice(b"\x1b[?7l");
        payload.extend_from_slice(&[b'x'; 85]);
        let args = make_parser_test_args(&payload);
        let shared = Arc::clone(&args.shared_parser);

        parser_read_loop(args);

        let parser = shared.lock().unwrap();
        let screen = parser.screen();
        assert_eq!(
            screen.cursor_position(),
            (0, 79),
            "DECAWM off must clamp the cursor to the right margin"
        );
        assert_eq!(
            screen.cell(0, 79).unwrap().contents(),
            "x",
            "the 85th char must overwrite the margin cell, not wrap"
        );
        assert!(
            (0..80).all(|col| screen
                .cell(1, col)
                .is_none_or(|cell| cell.contents().is_empty())),
            "row 1 must remain empty (no wrap)"
        );
    }

    /// Regression: pico inserts characters mid-line using IRM insert mode
    /// (`CSI 4 h` ... char ... `CSI 4 l`). If the emulator ignores insert mode,
    /// the char OVERWRITES the existing cell instead of shifting the row right
    /// ("types over existing characters" on a wrapped line). Replay pico's
    /// exact insert sequence through the production ingestion loop.
    #[test]
    fn parser_read_loop_insert_mode_inserts() {
        // "hello", move to col 2 (1-based) → (0,1) 0-based, enable IRM, insert 'X'.
        let args = make_parser_test_args(b"hello\x1b[1;2H\x1b[4hX\x1b[4l");
        let shared = Arc::clone(&args.shared_parser);

        parser_read_loop(args);

        let parser = shared.lock().unwrap();
        let screen = parser.screen();
        assert_eq!(screen.cell(0, 1).unwrap().contents(), "X", "inserted char");
        assert_eq!(
            screen.cell(0, 2).unwrap().contents(),
            "e",
            "existing text must shift right, not be overwritten"
        );
        assert_eq!(
            screen.cell(0, 5).unwrap().contents(),
            "o",
            "row tail preserved"
        );
    }

    #[test]
    fn parser_read_loop_status_callback_called_when_set() {
        let args = make_parser_test_args(b"data");
        let woke = Arc::new(AtomicBool::new(false));
        let woke_clone = Arc::clone(&woke);
        if let Ok(mut guard) = args.status_cb.lock() {
            *guard = Some(Box::new(move |status| {
                if status == crate::PtyStatus::Wakeup {
                    woke_clone.store(true, Ordering::Relaxed);
                }
            }));
        }

        parser_read_loop(args);

        assert!(
            woke.load(Ordering::Relaxed),
            "status callback must be invoked on wakeup"
        );
    }

    #[test]
    fn parser_read_loop_tracks_tail_for_cross_boundary_dsr() {
        let args = make_parser_test_args(b"XX\x1b[6nYY");
        let dsr_requested = Arc::clone(&args.dsr_requested);

        parser_read_loop(args);

        assert!(
            dsr_requested.load(Ordering::Relaxed),
            "DSR in combined data must be detected"
        );
    }

    #[test]
    fn parser_read_loop_relays_osc52_to_isolated_clipboard_and_hook() {
        // Inject a headless clipboard backed by an isolated shared in-memory
        // buffer so the relay is deterministic and the real system clipboard
        // / process-global default buffer are never touched.
        let shared = Arc::new(RwLock::new(None));
        let mut args = make_parser_test_args(&term_clipboard::format_osc52_bytes("clip via pty"));
        args.clipboard = Some(Clipboard::with_shared_buffer(Arc::clone(&shared)));
        let captured = Arc::new(Mutex::new(None));
        args.osc52_text = Some(Arc::clone(&captured));

        parser_read_loop(args);

        let captured = captured.lock().unwrap();
        assert_eq!(
            captured.as_deref(),
            Some("clip via pty"),
            "osc52_text hook must capture the relayed payload"
        );
        // The shared buffer must hold the relayed text, proving set() ran
        // without touching the real system clipboard.
        assert_eq!(
            *shared.read().unwrap(),
            Some("clip via pty".to_string()),
            "relayed set() must have written the shared in-memory buffer"
        );
    }

    #[test]
    fn set_status_callback_fires_from_spawn() {
        // Use cat, which blocks on input, so we control when output happens.
        // Portability: `cat` exists on Unix; on Windows the test is skipped.
        let cmd = CommandBuilder::new(get_test_executable());
        let size = PtySize {
            rows: 24,
            cols: 80,
            pixel_width: 0,
            pixel_height: 0,
        };
        let mut pty = Pty::spawn_with_scrollback(cmd, size, 100).expect("spawn_with_scrollback");

        let woke = Arc::new(AtomicBool::new(false));
        let woke_cb = Arc::clone(&woke);
        pty.set_status_callback(Some(Box::new(move |status| {
            if status == crate::PtyStatus::Wakeup {
                woke_cb.store(true, Ordering::Relaxed);
            }
        })));

        // Write to the PTY — terminal echo triggers a read on the master
        // side, which the reader thread processes and fires the callback.
        let _ = pty.write_str("hello\n");

        // Wait for callback with timeout (up to 5s)
        for _ in 0..250 {
            if woke.load(Ordering::Relaxed) {
                break;
            }
            std::thread::sleep(std::time::Duration::from_millis(20));
        }

        // Clean up
        if let Some(child) = pty.child.as_mut() {
            let _ = child.kill();
        }

        assert!(
            woke.load(Ordering::Relaxed),
            "status callback must fire on Wakeup when PTY outputs data"
        );
    }

    #[test]
    #[cfg(windows)]
    fn spawn_assigns_job_object_containing_child() {
        // The Windows child must be contained in a Job Object so kill paths
        // terminate the whole tree (grandchildren included), not just the
        // session leader. We spawn the mock in `spawn_child` mode: it forks a
        // grandchild (another mock instance in `sleep` mode) and prints
        // `GRANDCHILD_PID:<n>` to stdout. We read that PID, confirm the
        // grandchild is alive, kill the child via `kill_child` (which calls
        // `TerminateJobObject`), and confirm the grandchild died too — the
        // whole-tree guarantee.
        let mock = get_mock_bin();
        let mut cmd = CommandBuilder::new(mock);
        cmd.arg("spawn_child");
        cmd.arg("60000");
        let size = PtySize {
            rows: 24,
            cols: 80,
            pixel_width: 0,
            pixel_height: 0,
        };
        let mut pty = Pty::spawn_with_scrollback(cmd, size, 100).expect("spawn_with_scrollback");

        assert!(
            pty.job.is_some(),
            "spawned child must be contained in a Job Object on Windows"
        );

        // Poll `screen()` — which answers the child's startup DSR
        // cursor-position request — while accumulating output until the
        // `GRANDCHILD_PID:` marker appears.
        let mut accumulated = Vec::new();
        let start = std::time::Instant::now();
        let grandchild = loop {
            pty.screen();
            accumulated.extend_from_slice(&pty.drain_pending());
            if let Some(idx) = accumulated
                .windows(b"GRANDCHILD_PID:".len())
                .position(|w| w == b"GRANDCHILD_PID:")
            {
                let rest = &accumulated[idx + b"GRANDCHILD_PID:".len()..];
                let pid_str: String = rest
                    .iter()
                    .take_while(|b| b.is_ascii_digit())
                    .map(|b| *b as char)
                    .collect();
                break pid_str
                    .parse()
                    .unwrap_or_else(|_| panic!("invalid grandchild PID in {rest:?}"));
            }
            assert!(
                start.elapsed() < std::time::Duration::from_secs(5),
                "mock spawn_child should print a grandchild PID; got {accumulated:?}"
            );
            std::thread::sleep(std::time::Duration::from_millis(20));
        };

        // The grandchild must be alive before the kill.
        assert!(
            term_session_mock::process_is_alive(grandchild),
            "grandchild {grandchild} should be alive before kill_child"
        );

        // kill_child must consume the job (its Drop then fires
        // KILL_ON_JOB_CLOSE for any stragglers) and reap the child.
        pty.kill_child().expect("kill_child");
        assert!(pty.job.is_none(), "job must be taken on kill_child");
        assert!(pty.child.is_none(), "child must be reaped on kill_child");
        assert!(pty.exited, "pty must be marked exited after kill_child");

        // The grandchild must die with the tree — the whole point of the
        // Job Object containment. Poll briefly since termination is async.
        let start = std::time::Instant::now();
        while term_session_mock::process_is_alive(grandchild) {
            assert!(
                start.elapsed() < std::time::Duration::from_secs(5),
                "grandchild {grandchild} should be dead after kill_child"
            );
            std::thread::sleep(std::time::Duration::from_millis(20));
        }
    }

    #[test]
    fn has_exited_fires_exited_callback_when_child_dies() {
        let cmd = CommandBuilder::new(get_test_executable());
        let size = PtySize {
            rows: 24,
            cols: 80,
            pixel_width: 0,
            pixel_height: 0,
        };
        let mut pty = Pty::spawn_with_scrollback(cmd, size, 100).expect("spawn_with_scrollback");

        let exited_fired = Arc::new(AtomicBool::new(false));
        let exited_cb = Arc::clone(&exited_fired);
        pty.set_status_callback(Some(Box::new(move |status| {
            if status == crate::PtyStatus::Exited {
                exited_cb.store(true, Ordering::Relaxed);
            }
        })));

        // Kill the child so try_wait returns Ok(Some(...))
        if let Some(child) = pty.child.as_mut() {
            let _ = child.kill();
        }

        // Wait for child to be reaped
        for _ in 0..250 {
            if pty.has_exited() {
                break;
            }
            std::thread::sleep(std::time::Duration::from_millis(20));
        }

        assert!(
            exited_fired.load(Ordering::Relaxed),
            "has_exited() must fire PtyStatus::Exited callback when child dies"
        );
    }

    #[test]
    fn has_exited_idempotent_after_child_exits() {
        let cmd = CommandBuilder::new(get_test_executable());
        let size = PtySize {
            rows: 24,
            cols: 80,
            pixel_width: 0,
            pixel_height: 0,
        };
        let mut pty = Pty::spawn_with_scrollback(cmd, size, 100).expect("spawn_with_scrollback");

        let exit_count = Arc::new(AtomicUsize::new(0));
        let count_cb = Arc::clone(&exit_count);
        pty.set_status_callback(Some(Box::new(move |status| {
            if status == crate::PtyStatus::Exited {
                count_cb.fetch_add(1, Ordering::Relaxed);
            }
        })));

        // Kill the child
        if let Some(child) = pty.child.as_mut() {
            let _ = child.kill();
        }

        // Wait for first has_exited to succeed
        for _ in 0..250 {
            if pty.has_exited() {
                break;
            }
            std::thread::sleep(std::time::Duration::from_millis(20));
        }

        // Record count after has_exited() first returned true
        let count_after_first = exit_count.load(Ordering::Relaxed);

        // Call has_exited() again — must NOT fire the callback
        assert!(pty.has_exited(), "second call must also return true");

        std::thread::sleep(std::time::Duration::from_millis(100));
        let count_after_second = exit_count.load(Ordering::Relaxed);
        assert_eq!(
            count_after_first, count_after_second,
            "has_exited() must not re-fire the Exited callback after returning true"
        );
    }

    // ── set_status_callback with pre-exited child ────────────────────

    #[test]
    fn set_status_callback_fires_when_child_already_exited() {
        let cmd = CommandBuilder::new(get_test_executable());
        let size = PtySize {
            rows: 24,
            cols: 80,
            pixel_width: 0,
            pixel_height: 0,
        };
        let mut pty = Pty::spawn_with_scrollback(cmd, size, 100).expect("spawn_with_scrollback");

        // Kill and reap the child process, consuming the latch via has_exited().
        if let Some(child) = pty.child.as_mut() {
            let _ = child.kill();
        }
        for _ in 0..250 {
            if pty.has_exited() {
                break;
            }
            std::thread::sleep(std::time::Duration::from_millis(20));
        }

        let exited_fired = Arc::new(AtomicBool::new(false));
        let exited_cb = Arc::clone(&exited_fired);

        // Register a new callback — must fire immediately (self.exited branch
        // with latch reset).
        pty.set_status_callback(Some(Box::new(move |status| {
            if status == PtyStatus::Exited {
                exited_cb.store(true, Ordering::Relaxed);
            }
        })));

        assert!(
            exited_fired.load(Ordering::Relaxed),
            "set_status_callback must fire PtyStatus::Exited when child already exited"
        );
    }

    #[test]
    fn set_status_callback_fires_via_try_wait_after_direct_kill() {
        let cmd = CommandBuilder::new(get_test_executable());
        let size = PtySize {
            rows: 24,
            cols: 80,
            pixel_width: 0,
            pixel_height: 0,
        };
        let mut pty = Pty::spawn_with_scrollback(cmd, size, 100).expect("spawn_with_scrollback");

        // Kill the child but do NOT call has_exited() — let the try_wait()
        // path in set_status_callback detect the exit.
        if let Some(child) = pty.child.as_mut() {
            let _ = child.kill();
        }
        for _ in 0..250 {
            if let Some(Ok(Some(_))) = pty.child.as_mut().map(|c| c.try_wait()) {
                break;
            }
            std::thread::sleep(std::time::Duration::from_millis(20));
        }

        let exited_fired = Arc::new(AtomicBool::new(false));
        let exited_cb = Arc::clone(&exited_fired);

        pty.set_status_callback(Some(Box::new(move |status| {
            if status == PtyStatus::Exited {
                exited_cb.store(true, Ordering::Relaxed);
            }
        })));

        assert!(
            exited_fired.load(Ordering::Relaxed),
            "set_status_callback must fire PtyStatus::Exited via try_wait() when child killed"
        );
    }

    // ── screen / set_scrollback / into_parts / Drop ─────────────────
    //
    // These tests exercise the shared-parser screen sharing path (sync
    // with dirty=true), the set_scrollback mutation path, the into_parts()
    // shutdown signaling, and the Drop impl.  They use a real Pty spawned
    // with `cat` so the reader thread is alive.

    #[test]
    fn screen_loads_from_shared_parser_when_dirty() {
        let cmd = CommandBuilder::new(get_test_executable());
        let size = PtySize {
            rows: 24,
            cols: 80,
            pixel_width: 0,
            pixel_height: 0,
        };
        let mut pty = Pty::spawn_with_scrollback(cmd, size, 100).expect("spawn_with_scrollback");

        // Initially dirty is false — sync clears it.
        pty.screen();

        // Simulate reader thread publishing a new screen via shared parser.
        let mut new_parser = term_wm_vt100::Parser::new(24, 80, 100);
        new_parser.process(b"hello world");
        {
            let mut shared = pty.shared_parser.lock().unwrap();
            *shared = new_parser;
        }
        pty.dirty.store(true, Ordering::Release);

        // screen() should clear dirty.
        pty.screen();

        // Verify content is from the new screen by reading directly from the parser.
        {
            let parser = pty.shared_parser.lock().unwrap();
            if let Some(cell) = parser.screen().cell(0, 0) {
                let contents = cell.contents();
                assert!(
                    contents.contains('h'),
                    "expected 'h' from new screen, got {contents:?}"
                );
            }
        }

        assert!(!pty.dirty.load(Ordering::Acquire), "dirty must be cleared");

        // Clean up: kill child so the reader thread exits.
        if let Some(child) = pty.child.as_mut() {
            let _ = child.kill();
        }
    }

    #[test]
    fn screen_syncs_from_shared_parser() {
        let cmd = CommandBuilder::new(get_test_executable());
        let size = PtySize {
            rows: 24,
            cols: 80,
            pixel_width: 0,
            pixel_height: 0,
        };
        let mut pty = Pty::spawn_with_scrollback(cmd, size, 100).expect("spawn_with_scrollback");

        // Publish a new screen via shared parser.
        let mut new_parser = term_wm_vt100::Parser::new(24, 80, 100);
        new_parser.process(b"content");
        {
            let mut shared = pty.shared_parser.lock().unwrap();
            *shared = new_parser;
        }
        pty.dirty.store(true, Ordering::Release);

        // Sync dirty state.
        pty.screen();

        // Verify content from the new screen is accessible via the shared parser.
        {
            let parser = pty.shared_parser.lock().unwrap();
            let cell = parser.screen().cell(0, 0);
            assert!(cell.is_some(), "expected a cell at (0,0)");
            assert_eq!(cell.unwrap().contents(), "c");
        }

        // Clean up.
        if let Some(child) = pty.child.as_mut() {
            let _ = child.kill();
        }
    }

    #[test]
    fn set_scrollback_mutation_visible_through_scrollback_and_screen() {
        // Regression test: set_scrollback must mutate the shared parser's screen,
        // and scrollback() must read from the same parser.
        //
        // Generate enough output (30 lines in a 24-row terminal) to fill the
        // scrollback buffer so that set_scrollback(N) isn't clamped to 0.
        let cmd = CommandBuilder::new(get_test_executable());
        let size = PtySize {
            rows: 24,
            cols: 80,
            pixel_width: 0,
            pixel_height: 0,
        };
        let mut pty = Pty::spawn_with_scrollback(cmd, size, 100).expect("spawn_with_scrollback");

        // Consume any initial dirty.
        pty.screen();

        // Publish a screen with enough content to fill scrollback.
        // 30 lines in a 24-row terminal → 6 lines in the scrollback buffer.
        let mut lines = Vec::new();
        for i in 0..30 {
            writeln!(lines, "line {}", i).unwrap();
        }
        let mut parser = term_wm_vt100::Parser::new(24, 80, 100);
        parser.process(&lines);
        {
            let mut shared = pty.shared_parser.lock().unwrap();
            *shared = parser;
        }
        pty.dirty.store(true, Ordering::Release);

        // Sync.
        pty.screen();

        // Start at bottom (scrollback == 0).
        assert_eq!(pty.scrollback(), 0);

        // set_scrollback mutates the shared parser's screen directly.
        let sb_available = pty.max_scrollback();
        assert!(
            sb_available >= 3,
            "need at least 3 scrollback lines, got {sb_available}"
        );
        pty.set_scrollback(3);

        // scrollback() reads from the shared parser — must see the mutation.
        assert_eq!(
            pty.scrollback(),
            3,
            "scrollback() must reflect set_scrollback"
        );

        // Verify the shared parser's screen reflects the mutation.
        {
            let shared = pty.shared_parser.lock().unwrap();
            assert_eq!(
                shared.screen().scrollback(),
                3,
                "shared parser's screen must reflect set_scrollback"
            );
        }

        // Mutation survives subsequent sync calls.
        pty.screen();
        assert_eq!(
            pty.scrollback(),
            3,
            "mutation must survive repeated screen() calls without new data"
        );

        // New shared parser data replaces the mutation (expected).
        let mut parser2 = term_wm_vt100::Parser::new(24, 80, 100);
        parser2.process(b"fresh output");
        {
            let mut shared = pty.shared_parser.lock().unwrap();
            *shared = parser2;
        }
        pty.dirty.store(true, Ordering::Release);
        pty.screen();

        assert_eq!(
            pty.scrollback(),
            0,
            "new screen data must reset scrollback to its value"
        );

        // Clean up.
        if let Some(child) = pty.child.as_mut() {
            let _ = child.kill();
        }
    }

    #[test]
    fn set_scrollback_and_scrollback_consistent() {
        // set_scrollback() and scrollback() both operate on the shared parser.
        // Mutations via set_scrollback must be visible through scrollback().
        let cmd = CommandBuilder::new(get_test_executable());
        let size = PtySize {
            rows: 24,
            cols: 80,
            pixel_width: 0,
            pixel_height: 0,
        };
        let mut pty = Pty::spawn_with_scrollback(cmd, size, 100).expect("spawn_with_scrollback");

        // Sync.
        pty.screen();

        // Publish and load a screen with scrollback content (30 lines).
        let mut lines = Vec::new();
        for i in 0..30 {
            writeln!(lines, "line {}", i).unwrap();
        }
        let mut parser = term_wm_vt100::Parser::new(24, 80, 100);
        parser.process(&lines);
        {
            let mut shared = pty.shared_parser.lock().unwrap();
            *shared = parser;
        }
        pty.dirty.store(true, Ordering::Release);
        pty.screen();

        assert!(
            pty.max_scrollback() >= 3,
            "need enough scrollback for this test"
        );

        // Mutate via set_scrollback.
        pty.set_scrollback(3);

        // scrollback() must see the mutation.
        assert_eq!(
            pty.scrollback(),
            3,
            "scrollback() must see mutation made via set_scrollback"
        );

        // Mutate again.
        pty.set_scrollback(5);

        assert_eq!(
            pty.scrollback(),
            5,
            "scrollback() must see mutation made via set_scrollback"
        );

        // Clean up.
        if let Some(child) = pty.child.as_mut() {
            let _ = child.kill();
        }
    }

    #[test]
    fn into_parts_takes_child_and_reader() {
        let cmd = CommandBuilder::new(get_test_executable());
        let size = PtySize {
            rows: 24,
            cols: 80,
            pixel_width: 0,
            pixel_height: 0,
        };
        let mut pty = Pty::spawn_with_scrollback(cmd, size, 100).expect("spawn_with_scrollback");

        assert!(
            pty.reader_is_alive(),
            "reader should be alive before into_parts"
        );

        let parts = pty.into_parts();
        assert!(parts.child.is_some(), "child should be taken");
        assert!(
            parts.reader_handle.is_some(),
            "reader handle should be taken"
        );
        assert!(
            !pty.reader_is_alive(),
            "reader should be dead after into_parts"
        );
        assert!(pty.child.is_none(), "child should be None after into_parts");
    }

    #[test]
    fn set_status_callback_with_existing_reader_does_not_panic() {
        let cmd = CommandBuilder::new(get_test_executable());
        let size = PtySize {
            rows: 24,
            cols: 80,
            pixel_width: 0,
            pixel_height: 0,
        };
        let mut pty = Pty::spawn_with_scrollback(cmd, size, 100).expect("spawn_with_scrollback");

        pty.set_status_callback(Some(Box::new(|_| {})));

        // Also test clearing the callback.
        pty.set_status_callback(None);

        if let Some(child) = pty.child.as_mut() {
            let _ = child.kill();
        }
    }

    // ── wrap_err ────────────────────────────────────────────────────

    #[test]
    fn wrap_err_with_string() {
        let e = wrap_err("openpty", "permission denied");
        let s = format!("{}", e);
        assert!(s.contains("pty openpty failed: permission denied"));
    }

    #[test]
    fn wrap_err_with_io_error() {
        let io_err = io::Error::new(io::ErrorKind::NotFound, "file not found");
        let e = wrap_err("resize", io_err);
        let s = format!("{}", e);
        assert!(s.contains("pty resize failed"));
        assert!(s.contains("file not found"));
    }

    #[test]
    fn wrap_err_with_integer() {
        let e = wrap_err("spawn_command", 42);
        let s = format!("{}", e);
        assert!(s.contains("pty spawn_command failed: 42"));
    }

    #[test]
    fn take_pending_title_clones_foreground_not_consumes() {
        let cmd = CommandBuilder::new(get_test_executable());
        let size = PtySize {
            rows: 24,
            cols: 80,
            pixel_width: 0,
            pixel_height: 0,
        };
        let mut pty = Pty::spawn_with_scrollback(cmd, size, 100).expect("spawn");

        *pty.foreground_title.lock().unwrap() = Some("vim".to_string());

        assert_eq!(pty.take_pending_title(), Some("vim".to_string()));
        assert_eq!(pty.take_pending_title(), Some("vim".to_string()));

        if let Some(child) = pty.child.as_mut() {
            let _ = child.kill();
        }
    }

    #[test]
    fn take_pending_title_purges_stale_osc_when_fg_present() {
        let cmd = CommandBuilder::new(get_test_executable());
        let size = PtySize {
            rows: 24,
            cols: 80,
            pixel_width: 0,
            pixel_height: 0,
        };
        let mut pty = Pty::spawn_with_scrollback(cmd, size, 100).expect("spawn");

        *pty.foreground_title.lock().unwrap() = Some("vim".to_string());
        *pty.pending_title.lock().unwrap() = Some("user@host".to_string());

        assert_eq!(pty.take_pending_title(), Some("vim".to_string()));
        assert_eq!(
            *pty.pending_title.lock().unwrap(),
            None,
            "stale OSC title must be purged"
        );

        if let Some(child) = pty.child.as_mut() {
            let _ = child.kill();
        }
    }

    #[test]
    fn take_pending_title_falls_back_to_osc_when_no_fg() {
        let cmd = CommandBuilder::new(get_test_executable());
        let size = PtySize {
            rows: 24,
            cols: 80,
            pixel_width: 0,
            pixel_height: 0,
        };
        let mut pty = Pty::spawn_with_scrollback(cmd, size, 100).expect("spawn");

        *pty.foreground_title.lock().unwrap() = None;
        *pty.pending_title.lock().unwrap() = Some("user@host".to_string());

        assert_eq!(pty.take_pending_title(), Some("user@host".to_string()));
        assert_eq!(
            *pty.pending_title.lock().unwrap(),
            None,
            "OSC title must be consumed"
        );

        if let Some(child) = pty.child.as_mut() {
            let _ = child.kill();
        }
    }

    #[test]
    fn take_pending_title_returns_none_when_both_empty() {
        let cmd = CommandBuilder::new(get_test_executable());
        let size = PtySize {
            rows: 24,
            cols: 80,
            pixel_width: 0,
            pixel_height: 0,
        };
        let mut pty = Pty::spawn_with_scrollback(cmd, size, 100).expect("spawn");

        *pty.foreground_title.lock().unwrap() = None;
        *pty.pending_title.lock().unwrap() = None;

        assert_eq!(pty.take_pending_title(), None);

        if let Some(child) = pty.child.as_mut() {
            let _ = child.kill();
        }
    }

    // ── resize / set_size ──────────────────────────────────────────

    #[test]
    fn history_replay_corrupts_percent_lines_via_ansi_cursor_movements() {
        // The old resize replayed ALL accumulated history through a new
        // parser at the smaller width.  The raw PTY byte stream from a real
        // session contains ANSI escape sequences (cursor positioning, clear
        // screen, scroll regions) that were generated at the old width.
        //
        // When replayed at a smaller width, long lines wrap to multiple rows,
        // shifting everything after them.  Absolute cursor positioning that
        // was correct at the old width now lands on wrong rows, causing the
        // SAME `%` character to appear at multiple screen positions.
        // set_size avoids this by operating on the existing cell grid.

        // Simulate command output: each iteration writes a long line (with
        // `%`) at a specific absolute row using \x1b[row;colH.  At 80 cols
        // each line fits on one row.  At 30 cols each line wraps to ~3 rows,
        // so the absolute row positions overlap with prior wrapped content.
        let mut history = Vec::new();
        for i in 0..5 {
            // Place a 65-char line with `%` at absolute row i+1.
            // When replayed at 30 cols this wraps to 3 rows,
            // pushing subsequent content down.
            writeln!(history, "\x1b[{};1Hline {}: {} %", i + 1, i, "x".repeat(52)).unwrap();
        }

        // ── OLD: create parser at 30 cols, replay all history ──
        let mut old = term_wm_vt100::Parser::new(24, 30, 100);
        old.process(&history);
        let old_text = old.screen().contents();

        // ── NEW: process at 80 cols, then set_size to 30 ──
        let mut new = term_wm_vt100::Parser::new(24, 80, 100);
        new.process(&history);
        new.screen_mut().set_size(24, 30);
        let new_text = new.screen().contents();

        // History-replay re-interprets cursor positioning at 30 cols where
        // each line wraps to ~3 rows.  Absolute references like \x1b[2;1H
        // land inside wrapped continuations, jamming all 5 lines into 1 row.
        assert_eq!(
            old_text.lines().count(),
            1,
            "history-replay must collapse all lines into one row"
        );
        assert_eq!(
            new_text.lines().count(),
            5,
            "set_size must preserve all 5 logical lines (via reflow)"
        );
    }

    #[test]
    fn reflow_preserves_scrollback_content_on_width_shrink() {
        // Reflow must preserve previously-buffered scrollback content when the
        // terminal width shrinks (the non-alt-screen resize case), instead of
        // truncating the wrapped rows of a long line.
        //
        // The reflow logic lives in the `term-wm-vt100` fork, whose own test
        // suite (including `reflow_preserves_scrollback_content`) runs in that
        // fork's repository, not here. This test is a local regression guard
        // against a future version of the fork losing the behavior.
        let mut parser = term_wm_vt100::Parser::new(24, 80, 2000);
        let long_line = "x".repeat(200);
        for i in 0..40 {
            parser.process(format!("line {i}: {}\r\n", long_line).as_bytes());
        }

        // Count every written character across the whole scrollback + screen.
        let count_chars =
            |parser: &mut term_wm_vt100::Parser| -> std::collections::BTreeMap<char, usize> {
                let (rows, cols) = parser.screen().size();
                parser.screen_mut().set_scrollback(usize::MAX);
                let sb = parser.screen().scrollback();
                let mut map = std::collections::BTreeMap::new();
                for i in 0..sb {
                    parser.screen_mut().set_scrollback(sb - i);
                    for c in 0..cols {
                        if let Some(cell) = parser.screen().cell(0, c) {
                            for ch in cell.contents().chars() {
                                *map.entry(ch).or_insert(0) += 1;
                            }
                        }
                    }
                }
                parser.screen_mut().set_scrollback(0);
                for r in 0..rows {
                    for c in 0..cols {
                        if let Some(cell) = parser.screen().cell(r, c) {
                            for ch in cell.contents().chars() {
                                *map.entry(ch).or_insert(0) += 1;
                            }
                        }
                    }
                }
                parser.screen_mut().set_scrollback(0);
                map
            };

        let before = count_chars(&mut parser);
        parser.screen_mut().set_size(24, 40);
        let after = count_chars(&mut parser);

        assert_eq!(
            before, after,
            "width shrink must not lose scrollback content"
        );
    }

    #[test]
    fn test_exited_callback_atomic_latch_under_contention() {
        use std::sync::Arc;
        use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
        use std::thread;

        let call_count = Arc::new(AtomicUsize::new(0));
        let count_clone = Arc::clone(&call_count);

        let cb: StatusCallback = Arc::new(std::sync::Mutex::new(Some(Box::new(move |status| {
            if matches!(status, crate::PtyStatus::Exited) {
                count_clone.fetch_add(1, Ordering::SeqCst);
            }
        }))));

        let exited_emitted = Arc::new(AtomicBool::new(false));

        let handles: Vec<_> = (0..10)
            .map(|_| {
                let cb = Arc::clone(&cb);
                let emitted = Arc::clone(&exited_emitted);
                thread::spawn(move || {
                    if !emitted.swap(true, Ordering::AcqRel)
                        && let Ok(guard) = cb.lock()
                        && let Some(ref f) = *guard
                    {
                        f(crate::PtyStatus::Exited);
                    }
                })
            })
            .collect();

        for h in handles {
            h.join().unwrap();
        }

        assert_eq!(
            call_count.load(Ordering::SeqCst),
            1,
            "Exited callback must execute exactly once under thread contention"
        );
    }

    #[test]
    fn cursor_bounded_shrink_preserves_bottom_when_cursor_at_bottom() {
        let mut parser = term_wm_vt100::Parser::new(30, 80, 200);
        for i in 0..30 {
            parser.process(format!("line {}\r\n", i).as_bytes());
        }
        parser.process(b"LASTLINE");

        let (cursor_row, _) = parser.screen().cursor_position();
        let new_rows: u16 = 24;
        assert_eq!(cursor_row, 29, "cursor at bottom");

        let scroll_lines = cursor_row - new_rows + 1;
        assert_eq!(scroll_lines, 6);
        parser.process(format!("\x1b[{}S", scroll_lines).as_bytes());
        parser.screen_mut().set_size(new_rows, 80);

        assert!(
            parser.screen().contents().contains("LASTLINE"),
            "bottom content preserved when cursor at bottom"
        );
        assert_eq!(parser.screen().size(), (24, 80));
    }

    #[test]
    fn cursor_bounded_shrink_skips_when_cursor_above_new_height() {
        let mut parser = term_wm_vt100::Parser::new(30, 80, 200);
        for i in 0..10 {
            parser.process(format!("line {}\r\n", i).as_bytes());
        }
        parser.process(b"MIDLINE");

        let (cursor_row, _) = parser.screen().cursor_position();
        assert!(
            cursor_row < 24,
            "cursor ({cursor_row}) above new viewport height"
        );

        parser.screen_mut().set_size(24, 80);
        assert!(
            parser.screen().contents().contains("MIDLINE"),
            "content preserved without any SU shift"
        );
        assert_eq!(parser.screen().size(), (24, 80));
    }

    // ── Environment sanitization ─────────────────────────────────────

    fn fresh_cmd() -> CommandBuilder {
        CommandBuilder::new("true")
    }

    #[test]
    fn sanitize_sets_multiplexer_safe_term() {
        let mut cmd = fresh_cmd();
        sanitize_child_environment_with_lc_ctype(&mut cmd, None);
        assert_eq!(
            cmd.get_env("TERM").and_then(|v| v.to_str()),
            Some(expected_child_term())
        );
    }

    #[test]
    fn sanitize_sets_truecolor() {
        let mut cmd = fresh_cmd();
        sanitize_child_environment_with_lc_ctype(&mut cmd, None);
        assert_eq!(
            cmd.get_env("COLORTERM").and_then(|v| v.to_str()),
            Some("truecolor")
        );
    }

    #[test]
    fn sanitize_strips_invalid_lc_ctype_utf8() {
        let mut cmd = fresh_cmd();
        // Simulate the invalid `LC_CTYPE=UTF-8` macOS/OrbStack SSH hop injects.
        cmd.env("LC_CTYPE", "UTF-8");
        sanitize_child_environment_with_lc_ctype(&mut cmd, Some("UTF-8"));
        assert!(
            cmd.get_env("LC_CTYPE").is_none(),
            "invalid LC_CTYPE=UTF-8 must be stripped"
        );
    }

    #[test]
    fn sanitize_keeps_valid_lc_ctype() {
        let mut cmd = fresh_cmd();
        cmd.env("LC_CTYPE", "en_US.UTF-8");
        sanitize_child_environment_with_lc_ctype(&mut cmd, Some("en_US.UTF-8"));
        assert_eq!(
            cmd.get_env("LC_CTYPE").and_then(|v| v.to_str()),
            Some("en_US.UTF-8"),
            "valid locale must be preserved"
        );
    }

    /// End-to-end: the child spawned via the real `Pty` launcher must see the
    /// sanitized environment. We write a child that prints its `TERM`,
    /// `COLORTERM`, and `LC_CTYPE` (or its absence) and assert on the output.
    #[test]
    #[cfg(unix)]
    fn spawned_child_receives_sanitized_environment() {
        // `sh` is POSIX-standard; on Windows this test is skipped.
        let script = indoc::indoc! {r#"
            TERM_VAL="${TERM:-<unset>}"
            COLORTERM_VAL="${COLORTERM:-<unset>}"
            LC_CTYPE_VAL="${LC_CTYPE:-<unset>}"
            printf 'TERM=%s\nCOLORTERM=%s\nLC_CTYPE=%s\n' \
              "$TERM_VAL" "$COLORTERM_VAL" "$LC_CTYPE_VAL"
        "#};
        let mut cmd = CommandBuilder::new("sh");
        cmd.arg("-c");
        cmd.arg(script);
        let size = PtySize {
            rows: 24,
            cols: 80,
            pixel_width: 0,
            pixel_height: 0,
        };
        let mut pty = Pty::spawn_with_scrollback(cmd, size, 0).expect("spawn");

        let mut accumulated = Vec::new();
        let start = std::time::Instant::now();
        loop {
            pty.screen();
            accumulated.extend_from_slice(&pty.drain_pending());
            let out = String::from_utf8_lossy(&accumulated);
            if out.contains("LC_CTYPE=") {
                break;
            }
            assert!(
                start.elapsed() < std::time::Duration::from_secs(5),
                "child never printed sanitized env; got {out}"
            );
            std::thread::sleep(std::time::Duration::from_millis(20));
        }
        if let Some(child) = pty.child.as_mut() {
            let _ = child.kill();
        }

        let out = String::from_utf8_lossy(&accumulated);
        assert!(
            out.contains(&format!("TERM={}", expected_child_term())),
            "expected TERM={}, got {out}",
            expected_child_term()
        );
        assert!(
            out.contains("COLORTERM=truecolor"),
            "expected COLORTERM=truecolor, got {out}"
        );
        assert!(
            !out.contains("LC_CTYPE=UTF-8"),
            "invalid LC_CTYPE=UTF-8 must not reach the child, got {out}"
        );
    }

    /// Mirrors the platform-specific `CHILD_TERM` so tests assert the value the
    /// sanitizer actually applies on this platform.
    fn expected_child_term() -> &'static str {
        CHILD_TERM
    }
}