term-wm-pty-engine 0.9.4-alpha

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

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

use crate::clipboard::{Clipboard, Osc52Extractor};
use crate::pty_state_tracker::PtyPerformAdapter;

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

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

/// 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 {
    master: 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<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)>,
    size: PtySize,
    pty_size: PtySize,
    scrollback_len: usize,
    child: Option<Box<dyn Child + Send + Sync>>,
    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<()>>,
}

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

    pub fn spawn_with_scrollback(
        command: CommandBuilder,
        size: PtySize,
        scrollback_len: usize,
    ) -> PtyResult<Self> {
        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))?;
        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))?,
        );
        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 = 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_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);
        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,
            })
        });
        Ok(Self {
            master: pair.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,
            tracker,
            size,
            pty_size: size,
            scrollback_len,
            child: Some(child),
            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();
        }
        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()
    }

    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(());
        }
        if size == self.pty_size {
            return Ok(());
        }
        // Lock the shared parser before resizing the OS PTY so the reader
        // thread cannot process post-SIGWINCH bytes against stale dimensions.
        let sp = self.shared_parser.clone();
        let mut guard = sp.lock().unwrap();
        let old_rows = self.pty_size.rows;
        let new_rows = size.rows;
        if new_rows < old_rows && !self.tracker.has_custom_margins() {
            let (cursor_row, _) = guard.screen().cursor_position();
            if cursor_row >= new_rows {
                let scroll_lines = cursor_row - new_rows + 1;
                let seq = format!("[{scroll_lines}S");
                guard.process(seq.as_bytes());
            }
        }
        self.master
            .resize(size)
            .map_err(|err| wrap_err("resize", err))?;
        guard.screen_mut().set_size(size.rows, size.cols);
        drop(guard);
        self.tracker.resize(size.rows);
        self.pty_size = size;
        self.size = size;
        Ok(())
    }

    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.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)
    }

    pub fn screen_lines(&mut self) -> Vec<String> {
        self.screen(); // sync dirty state
        let parser = self.shared_parser.lock().unwrap();
        let screen = parser.screen();
        let contents = screen.contents();
        let mut lines: Vec<String> = contents.lines().map(|line| line.to_string()).collect();
        if lines.len() < self.size.rows as usize {
            lines.resize(self.size.rows as usize, 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.
    pub fn kill_child(&mut self) -> PtyResult<()> {
        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.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.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
    }

    /// 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();
        let parser = self.shared_parser.lock().unwrap();
        parser.screen().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();
        }
    }
}

/// 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<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>,
}

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,
    } = 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 final payload of a burst is never dropped.
    let mut clipboard = clipboard.unwrap_or_else(Clipboard::new);
    const IO_BURST_BUDGET: usize = 256 * 1024; // 256 KB
    loop {
        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() {
                    clipboard.set(&text);
                    if let Some(ref capture) = osc52_text {
                        *capture.lock().unwrap() = Some(text);
                    }
                }
                // Send wakeup for final screen, then exited.
                if let Ok(guard) = status_cb.lock()
                    && let Some(ref cb) = *guard
                {
                    cb(crate::PtyStatus::Wakeup);
                    if !exited_emitted.swap(true, Ordering::AcqRel) {
                        cb(crate::PtyStatus::Exited);
                    }
                }
                break;
            }
            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);
                }
                if let Ok(mut last) = last_bytes.lock() {
                    last.clear();
                    last.extend_from_slice(&buf[..n]);
                }
                if let Ok(mut p) = pending.lock() {
                    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_routing = tracker.requires_app_routing();
                vte_parser.advance(&mut tracker_adapter, &buf[..n]);
                let new_routing = tracker.requires_app_routing();
                if prev_routing != new_routing {
                    tracing::info!(
                        "[STAGE 1] PTY routing flipped: {} -> {}",
                        prev_routing,
                        new_routing
                    );
                    if let Ok(guard) = status_cb.lock()
                        && let Some(ref cb) = *guard
                    {
                        cb(crate::PtyStatus::DirectInputChanged(new_routing));
                    } 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();
                    shared.process(&buf[..n]);
                }

                if let Some(title) = extract_osc_title(&buf[..n])
                    && let Ok(mut guard) = pending_title.lock()
                {
                    *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.
                if let Some(text) = osc52.push(&buf[..n], &prev_tail) {
                    clipboard.set(&text);
                    if let Some(ref capture) = osc52_text {
                        *capture.lock().unwrap() = 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) {
                    if let Ok(guard) = status_cb.lock()
                        && 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();
                    while dirty.load(Ordering::Acquire) {
                        guard = cvar.wait(guard).unwrap();
                    }
                    bytes_since_render = 0;
                }

                // 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(_) => {
                if let Ok(guard) = status_cb.lock()
                    && let Some(ref cb) = *guard
                    && !exited_emitted.swap(true, Ordering::AcqRel)
                {
                    cb(crate::PtyStatus::Exited);
                }
                break;
            }
        }
    }
}

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;
    }
}

#[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;
    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"
        }
    }

    // ── 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() -> ParserReadLoopArgs {
        ParserReadLoopArgs {
            reader: Box::new(Cursor::new(Vec::new())),
            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(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)),
        }
    }

    #[test]
    fn parser_read_loop_reads_and_sets_pending_and_last() {
        let payload = b"hello\r\n\x1b[6nworld";
        let mut args = make_parser_test_args();
        args.reader = Box::new(Cursor::new(payload.to_vec()));
        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 mut args = make_parser_test_args();
        args.reader = Box::new(Cursor::new(Vec::new()));
        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));
    }

    #[test]
    fn parser_read_loop_status_callback_called_when_set() {
        let payload = b"data";
        let mut args = make_parser_test_args();
        args.reader = Box::new(Cursor::new(payload.to_vec()));
        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 payload = b"XX\x1b[6nYY";
        let mut args = make_parser_test_args();
        args.reader = Box::new(Cursor::new(payload.to_vec()));
        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();
        args.clipboard = Some(Clipboard::with_shared_buffer(Arc::clone(&shared)));
        let captured = Arc::new(Mutex::new(None));
        args.osc52_text = Some(Arc::clone(&captured));
        args.reader = Box::new(Cursor::new(crate::clipboard::format_osc52_bytes(
            "clip via pty",
        )));

        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.
        // TODO: add Windows support with `cmd /c type CON`
        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]
    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 = 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 = 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 = 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 = 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 = 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 = 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 = 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 5 separate rows"
        );
    }

    #[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 = 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 = 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));
    }
}