tastty 0.1.0

//! Session lifecycle and control types.
//!
//! This module owns spawning, I/O callbacks, and the managed/unmanaged
//! session markers.

use std::fmt;
use std::io::{Read, Write};
use std::sync::atomic::{AtomicBool, AtomicU64};
use std::sync::{Arc, RwLock};
use std::thread::JoinHandle;
use std::time::Instant;

use bytes::Bytes;
use tastty_core::frame::{bracketed_paste, focus_report};
use tastty_core::{
    KeyEncoder, KeyEvent, MouseEncoder, MouseEvent, Parser, Screen, ScreenEvent, TerminalMode,
    TerminalSize,
};
use tokio::sync::mpsc;
use tracing::{debug, debug_span, warn};

use crate::error::{Error, IoError, Result};
use crate::process_group;

mod io;
mod managed;
mod options;
mod unmanaged;

pub use io::WriterHandle;
pub use options::{KeyAction, SessionOptions};

use io::{JOIN_TIMEOUT, ReaderContext, join_with_timeout, spawn_reader, spawn_writer};
use managed::{ManagedState, SHUTDOWN_TIMEOUT, poll_exit, wait_for_exit_or_deadline};
use options::KeyCallback;

mod sealed {
    pub trait Sealed {}
}

/// Marker trait for the session ownership mode.
pub trait TerminalOwnership: sealed::Sealed {}

/// Marker type for a session that owns the child process.
pub struct Managed;
/// Marker type for a session that is detached from child ownership.
pub struct Unmanaged;

impl sealed::Sealed for Managed {}
impl sealed::Sealed for Unmanaged {}
impl TerminalOwnership for Managed {}
impl TerminalOwnership for Unmanaged {}

/// A [`Terminal`] that owns and reaps its child process.
pub type ManagedTerminal = Terminal<Managed>;
/// A [`Terminal`] that does not own child process lifecycle.
pub type UnmanagedTerminal = Terminal<Unmanaged>;

/// A live PTY session with async I/O, screen parsing, and resize support.
///
/// The reader thread drives the inner [`tastty_core::Parser`] and routes
/// every drained query [`ScreenEvent`] ([DA1/DA2/DA3, DSR, CPR, DECRQM,
/// DECRQSS, XTGETTCAP, OSC color queries, XTWINOPS][xterm-ctlseqs], [Kitty
/// keyboard][kitty-kbd]) through
/// [`tastty_core::host_reply::auto_reply_bytes`] back to the child on every
/// read tick. Embedders do not need to handle replies manually unless they
/// want to override per-query behavior; see
/// [`SessionOptions::on_host_query`] for the override hook.
///
/// Dropping the handle terminates the child and joins the I/O threads with
/// bounded blocking; see the [`Drop`] impl for the timing breakdown and
/// hot-path mitigations.
///
/// [xterm-ctlseqs]: https://invisible-island.net/xterm/ctlseqs/ctlseqs.html
/// [kitty-kbd]: https://sw.kovidgoyal.net/kitty/keyboard-protocol/
pub struct Terminal<Mode: TerminalOwnership = Managed> {
    pub(super) parser: Arc<RwLock<Parser>>,
    pub(super) writer_tx: Option<mpsc::Sender<Bytes>>,
    pub(super) managed: Option<ManagedState>,
    pub(super) reader_handle: Option<std::thread::JoinHandle<()>>,
    pub(super) writer_handle: Option<std::thread::JoinHandle<()>>,
    pub(super) dirty: Arc<AtomicBool>,
    pub(super) redraw_notify: Arc<tokio::sync::Notify>,
    pub(super) event_rx: std::sync::Mutex<Option<mpsc::UnboundedReceiver<ScreenEvent>>>,
    pub(super) io_error_rx: std::sync::Mutex<Option<mpsc::UnboundedReceiver<IoError>>>,
    /// Milliseconds since `epoch` when bytes last arrived during sync mode.
    pub(super) last_sync_arrival_ms: Arc<AtomicU64>,
    pub(super) epoch: Instant,
    pub(super) virtual_cols: Option<u16>,
    pub(super) key_callback: Option<KeyCallback>,
    pub(super) _mode: std::marker::PhantomData<Mode>,
}

pub(super) struct CommonParts {
    pub(super) parser: Arc<RwLock<Parser>>,
    pub(super) writer_tx: mpsc::Sender<Bytes>,
    pub(super) reader_handle: std::thread::JoinHandle<()>,
    pub(super) writer_handle: std::thread::JoinHandle<()>,
    pub(super) dirty: Arc<AtomicBool>,
    pub(super) redraw_notify: Arc<tokio::sync::Notify>,
    pub(super) event_rx: mpsc::UnboundedReceiver<ScreenEvent>,
    pub(super) io_error_rx: mpsc::UnboundedReceiver<IoError>,
    pub(super) last_sync_arrival_ms: Arc<AtomicU64>,
    pub(super) epoch: Instant,
    pub(super) virtual_cols: Option<u16>,
    pub(super) key_callback: Option<KeyCallback>,
}

pub(super) fn build_common(
    reader: Box<dyn Read + Send>,
    writer: Box<dyn Write + Send>,
    opts: &mut SessionOptions,
    reader_done: Option<std::sync::mpsc::Sender<()>>,
) -> Result<CommonParts> {
    let pty_size = opts.pty_size();
    let parser_cols = opts.virtual_cols.unwrap_or(pty_size.cols);
    let span = debug_span!(
        "terminal_session_build",
        rows = pty_size.rows,
        cols = pty_size.cols,
        parser_cols,
        scrollback = opts.scrollback,
        has_output_callback = opts.output_callback.is_some(),
        has_input_callback = opts.input_callback.is_some(),
        has_redraw_callback = opts.redraw_callback.is_some(),
    );
    let _guard = span.enter();
    debug!("building terminal session internals");
    let parser_size = TerminalSize {
        rows: pty_size.rows,
        cols: parser_cols,
    };
    let host_profile = Arc::new(opts.host_profile.take().unwrap_or_default());
    let mut p = Parser::with_profile(
        parser_size,
        opts.scrollback as usize,
        Arc::clone(&host_profile),
    );
    if opts.pixel_cell_size.width > 0 {
        p.screen_mut().set_pixel_cell_size(opts.pixel_cell_size);
    }
    let parser = Arc::new(RwLock::new(p));
    debug!("parser initialized");

    let (io_error_tx, io_error_rx) = mpsc::unbounded_channel::<IoError>();
    let (writer_tx, writer_handle) =
        spawn_writer(writer, opts.input_callback.take(), io_error_tx.clone())?;
    debug!("writer thread spawned");

    let dirty = Arc::new(AtomicBool::new(false));
    let redraw_notify = Arc::new(tokio::sync::Notify::new());
    let epoch = Instant::now();
    let last_sync_arrival_ms = Arc::new(AtomicU64::new(0));

    let (event_tx, event_rx) = mpsc::unbounded_channel::<ScreenEvent>();
    let reader_handle = match spawn_reader(ReaderContext {
        reader,
        parser: Arc::clone(&parser),
        host_profile,
        response_tx: writer_tx.clone(),
        dirty: Arc::clone(&dirty),
        redraw_notify: Arc::clone(&redraw_notify),
        last_sync_arrival_ms: Arc::clone(&last_sync_arrival_ms),
        epoch,
        event_tx,
        io_error_tx,
        output_callback: opts.output_callback.take(),
        redraw_callback: opts.redraw_callback.take(),
        host_query_callback: Arc::clone(&opts.host_query_callback),
        clipboard_policy: opts.clipboard_policy,
        reader_done,
    }) {
        Ok(handle) => handle,
        Err(err) => {
            drop(writer_tx);
            join_with_timeout(Some(writer_handle), JOIN_TIMEOUT);
            return Err(err);
        }
    };
    debug!("reader thread spawned");

    Ok(CommonParts {
        parser,
        writer_tx,
        reader_handle,
        writer_handle,
        dirty,
        redraw_notify,
        event_rx,
        io_error_rx,
        last_sync_arrival_ms,
        epoch,
        virtual_cols: opts.virtual_cols,
        key_callback: opts.key_callback.take(),
    })
}

pub(super) fn spawn_named<F>(name: &'static str, f: F) -> Result<JoinHandle<()>>
where
    F: FnOnce() + Send + 'static,
{
    #[cfg(test)]
    if consume_thread_spawn_failure(name) {
        return Err(Error::ThreadSpawn {
            name,
            source: std::io::Error::other("injected thread spawn failure"),
        });
    }

    std::thread::Builder::new()
        .name(name.into())
        .spawn(f)
        .map_err(|source| Error::ThreadSpawn { name, source })
}

#[cfg(test)]
thread_local! {
    static THREAD_SPAWN_FAILURES: std::cell::RefCell<Vec<&'static str>> =
        const { std::cell::RefCell::new(Vec::new()) };
}

#[cfg(test)]
pub(super) fn fail_next_thread_spawn(name: &'static str) {
    THREAD_SPAWN_FAILURES.with(|failures| failures.borrow_mut().push(name));
}

#[cfg(test)]
fn consume_thread_spawn_failure(name: &'static str) -> bool {
    THREAD_SPAWN_FAILURES.with(|failures| {
        let mut failures = failures.borrow_mut();
        let Some(pos) = failures.iter().position(|candidate| *candidate == name) else {
            return false;
        };
        failures.remove(pos);
        true
    })
}

impl<M: TerminalOwnership> Terminal<M> {
    /// Acquire a read guard on the parser, recovering from poison so a
    /// prior panic while a guard was held is survivable instead of fatal.
    pub(super) fn parser_read(&self) -> std::sync::RwLockReadGuard<'_, Parser> {
        self.parser
            .read()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
    }

    /// Write-guard counterpart to [`parser_read`](Self::parser_read).
    pub(super) fn parser_write(&self) -> std::sync::RwLockWriteGuard<'_, Parser> {
        self.parser
            .write()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
    }

    /// Send a typed host reply to the child process.
    ///
    /// This is the preferred way to respond to host query
    /// [`ScreenEvent`] variants. The reply is encoded
    /// into the correct escape sequence automatically.
    ///
    /// # Errors
    ///
    /// Returns the same errors as [`send`](Self::send).
    ///
    /// ```no_run
    /// # use tastty::{HostReply, ScreenEvent};
    /// # fn handle(event: ScreenEvent, session: &tastty::Terminal) {
    /// if let ScreenEvent::Xtversion = event {
    ///     session.reply(HostReply::Xtversion("myterm(1.0)".into())).unwrap();
    /// }
    /// # }
    /// ```
    pub fn reply(&self, reply: crate::host_reply::HostReply) -> Result<()> {
        self.send(&reply.encode())
    }

    /// Like [`reply`](Self::reply) but waits asynchronously if the queue is full.
    ///
    /// # Cancellation safety
    ///
    /// Cancellation-safe; the reply is encoded synchronously and then
    /// handed to [`send_async`](Self::send_async), which is itself
    /// cancel-safe. A cancelled call leaves no partial reply in the
    /// writer queue.
    ///
    /// # Errors
    ///
    /// Returns the same errors as [`send_async`](Self::send_async).
    pub async fn reply_async(&self, reply: crate::host_reply::HostReply) -> Result<()> {
        self.send_async(&reply.encode()).await
    }

    /// Send raw bytes to the child process.
    ///
    /// Also used to inject responses into the PTY, such as clipboard
    /// read responses (OSC 52).
    ///
    /// # Errors
    ///
    /// Returns [`Error::SendQueueFull`] when the bounded writer queue is
    /// full, or [`Error::SendClosed`] when the writer side is gone.
    pub fn send(&self, bytes: &[u8]) -> Result<()> {
        let tx = self.writer_tx.as_ref().ok_or(Error::SendClosed)?;
        match tx.try_send(Bytes::copy_from_slice(bytes)) {
            Ok(()) => Ok(()),
            Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => Err(Error::SendQueueFull),
            Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => Err(Error::SendClosed),
        }
    }

    /// Encode and send a key event using the current terminal input mode.
    ///
    /// # Errors
    ///
    /// Returns the same errors as [`send`](Self::send).
    pub fn send_key(&self, key: impl Into<KeyEvent>) -> Result<()> {
        if let Some(bytes) = self.encode_key(key.into())? {
            self.send(&bytes)?;
        }
        Ok(())
    }

    /// Send bytes, blocking the current thread if the writer queue is full.
    ///
    /// Used internally for framed multi-byte sequences (bracketed paste,
    /// focus report) where a partial send would corrupt the child's parser
    /// state if the queue filled between pieces.
    ///
    /// Rejects callers already inside a tokio runtime with
    /// [`Error::BlockingInsideAsync`]: `tokio::mpsc::Sender::blocking_send`
    /// would panic there, and a typed error is friendlier than a panic for
    /// a preventable misuse.
    fn send_blocking(&self, bytes: &[u8]) -> Result<()> {
        if tokio::runtime::Handle::try_current().is_ok() {
            return Err(Error::BlockingInsideAsync);
        }
        let tx = self.writer_tx.as_ref().ok_or(Error::SendClosed)?;
        tx.blocking_send(Bytes::copy_from_slice(bytes))
            .map_err(|_err| Error::SendClosed)
    }

    /// Build a bracketed-paste frame for `text`, consulting the parser
    /// for whether the child enabled the markers.
    ///
    /// The parser read guard is released before this method returns;
    /// callers can hold the resulting `Vec<u8>` across an `.await`
    /// without retaining a guard, and outer locks (e.g. an enclosing
    /// `Mutex<Terminal>`) can be dropped after this call without
    /// affecting the frame contents.
    pub fn paste_frame(&self, text: &str) -> Vec<u8> {
        let bracketed = self
            .parser_read()
            .screen()
            .mode(TerminalMode::BracketedPaste);
        bracketed_paste(text, bracketed)
    }

    /// Return the focus-report byte sequence when the child has enabled
    /// focus reporting, or `None` when the mode is off and the call
    /// should be a no-op.
    ///
    /// As with [`paste_frame`](Self::paste_frame), the parser read
    /// guard is released before this method returns.
    pub fn focus_frame(&self, gained: bool) -> Option<&'static [u8]> {
        if self.parser_read().screen().mode(TerminalMode::FocusInOut) {
            Some(focus_report(gained))
        } else {
            None
        }
    }

    /// Clone the writer-side handle for off-lock async sends.
    ///
    /// Returns `None` once the writer thread has been torn down (after
    /// drop or a fatal writer error). Callers that hold an outer lock
    /// around this `Terminal` can use this to extract the handle
    /// synchronously, drop the lock, and then `.await` the send on the
    /// returned [`WriterHandle`] without holding the outer guard across
    /// the await point. The driver's `Session::send_paste_async` is the
    /// canonical consumer.
    #[must_use]
    pub fn writer(&self) -> Option<WriterHandle> {
        self.writer_tx
            .as_ref()
            .map(|tx| WriterHandle::new(tx.clone()))
    }

    /// Wrap `text` in bracketed-paste markers when the child has requested
    /// them, and deliver the sequence atomically.
    ///
    /// Markers and payload are concatenated into a single frame so that a
    /// single mpsc insert covers the whole sequence; concurrent callers
    /// cannot interleave `[200~` from one paste with `[201~` from another.
    /// Blocks if the writer queue is full. **Do not call from async code**;
    /// use `send_paste_async`.
    ///
    /// # Errors
    ///
    /// Returns [`Error::BlockingInsideAsync`] if called from within a Tokio
    /// runtime, or [`Error::SendClosed`] if the writer side is gone.
    pub fn send_paste(&self, text: &str) -> Result<()> {
        self.send_blocking(&self.paste_frame(text))
    }

    /// Deliver a focus-in or focus-out report when the child has enabled
    /// focus reporting.
    ///
    /// Blocks if the writer queue is full. **Do not call from async code**;
    /// use `send_focus_async`.
    ///
    /// # Errors
    ///
    /// Returns [`Error::BlockingInsideAsync`] if called from within a Tokio
    /// runtime, or [`Error::SendClosed`] if the writer side is gone.
    pub fn send_focus(&self, gained: bool) -> Result<()> {
        if let Some(bytes) = self.focus_frame(gained) {
            self.send_blocking(bytes)?;
        }
        Ok(())
    }

    /// Encode and send a mouse event if the child has enabled mouse reporting.
    ///
    /// # Errors
    ///
    /// Returns the same errors as [`send`](Self::send).
    pub fn send_mouse(&self, event: impl Into<MouseEvent>) -> Result<()> {
        let event = event.into();
        let enc = self.mouse_encoder();
        if let Some(bytes) = enc.encode_mouse(&event) {
            self.send(&bytes)?;
        }
        Ok(())
    }

    /// Send raw bytes, waiting asynchronously if the queue is full.
    /// Unlike [`send`](Self::send), this never returns `SendQueueFull`.
    ///
    /// # Cancellation safety
    ///
    /// Cancellation-safe; uses `tokio::sync::mpsc::Sender::send`
    /// internally, which is documented as cancel-safe by tokio.
    /// A cancelled send neither queues nor drops the value.
    ///
    /// # Errors
    ///
    /// Returns [`Error::SendClosed`] when the writer side is gone.
    pub async fn send_async(&self, bytes: &[u8]) -> Result<()> {
        let tx = self.writer_tx.as_ref().ok_or(Error::SendClosed)?;
        tx.send(Bytes::copy_from_slice(bytes))
            .await
            .map_err(|_err| Error::SendClosed)
    }

    /// Like [`send_key`](Self::send_key) but waits if the queue is full.
    ///
    /// # Cancellation safety
    ///
    /// Cancellation-safe; the key is encoded synchronously (under a
    /// short-lived parser read guard that is dropped before the
    /// `.await`) and the resulting bytes are forwarded to
    /// [`send_async`](Self::send_async), which is cancel-safe.
    ///
    /// # Errors
    ///
    /// Returns the same errors as [`send_async`](Self::send_async).
    pub async fn send_key_async(&self, key: impl Into<KeyEvent>) -> Result<()> {
        let encoded = self.encode_key(key.into())?;
        if let Some(bytes) = encoded {
            self.send_async(&bytes).await?;
        }
        Ok(())
    }

    /// Like [`send_paste`](Self::send_paste) but waits if the queue is full.
    ///
    /// The bracketed-paste markers and payload are concatenated into a
    /// single frame before sending, matching the atomicity guarantee of
    /// the sync variant.
    ///
    /// # Cancellation safety
    ///
    /// Cancellation-safe; the frame is built synchronously (the
    /// bracketed-paste mode probe takes and releases a parser read
    /// guard before the `.await`) and then handed to
    /// [`send_async`](Self::send_async), which is cancel-safe. Frame
    /// atomicity is preserved across cancellation: a cancelled call
    /// either delivered the entire frame or none of it.
    ///
    /// # Errors
    ///
    /// Returns [`Error::SendClosed`] when the writer side is gone.
    pub async fn send_paste_async(&self, text: &str) -> Result<()> {
        let frame = self.paste_frame(text);
        self.send_async(&frame).await
    }

    /// Like [`send_focus`](Self::send_focus) but waits if the queue is full.
    ///
    /// # Cancellation safety
    ///
    /// Cancellation-safe; the focus-report probe takes and releases a
    /// parser read guard synchronously before the `.await`, and the
    /// resulting bytes are forwarded to
    /// [`send_async`](Self::send_async), which is cancel-safe.
    ///
    /// # Errors
    ///
    /// Returns [`Error::SendClosed`] when the writer side is gone.
    pub async fn send_focus_async(&self, gained: bool) -> Result<()> {
        if let Some(bytes) = self.focus_frame(gained) {
            self.send_async(bytes).await?;
        }
        Ok(())
    }

    /// Create a [`KeyEncoder`] synced to the current terminal state.
    pub fn key_encoder(&self) -> KeyEncoder {
        let mut enc = KeyEncoder::new();
        let parser = self.parser_read();
        enc.sync(parser.screen());
        enc
    }

    /// Create a [`MouseEncoder`] synced to the current terminal state.
    pub fn mouse_encoder(&self) -> MouseEncoder {
        let mut enc = MouseEncoder::new();
        let parser = self.parser_read();
        enc.sync(parser.screen());
        enc
    }

    /// Encode a key event into bytes without sending. Returns `None` if
    /// no encoding is available for this key.
    fn encode_key(&self, key: KeyEvent) -> Result<Option<Vec<u8>>> {
        let enc = self.key_encoder();

        if let Some(ref cb) = self.key_callback {
            match cb(&key, *enc.screen_state()) {
                KeyAction::Drop => return Ok(None),
                KeyAction::Replace(bytes) => return Ok(Some(bytes)),
                KeyAction::Send => {}
            }
        }

        Ok(enc.encode_key(&key))
    }

    /// Access the screen without cloning. The closure receives a reference
    /// to the screen while the read lock is held; the lock is released
    /// when the closure returns.
    ///
    /// This is the canonical zero-copy accessor; per-frame render loops
    /// should call this rather than allocating a new `Screen` on every
    /// frame.
    pub fn with_screen<F, R>(&self, f: F) -> R
    where
        F: FnOnce(&Screen) -> R,
    {
        let parser = self.parser_read();
        f(parser.screen())
    }

    /// Return the current terminal size.
    #[must_use]
    pub fn size(&self) -> TerminalSize {
        self.with_screen(Screen::size)
    }

    /// Scroll the viewport up by `count` rows into scrollback history.
    pub fn scroll_up(&self, count: usize) {
        self.parser_write().screen_mut().scroll_up(count);
    }

    /// Scroll the viewport down by `count` rows (towards live output).
    pub fn scroll_down(&self, count: usize) {
        self.parser_write().screen_mut().scroll_down(count);
    }

    /// Reset the scrollback viewport to show live output.
    pub fn scroll_reset(&self) {
        self.parser_write().screen_mut().scroll_reset();
    }

    /// Set the scrollback viewport to an absolute offset.
    /// 0 = live output, scrollback_available() = oldest history.
    pub fn scroll_to(&self, offset: usize) {
        self.parser_write().screen_mut().scroll_to(offset);
    }

    /// Current viewport offset into scrollback history.
    ///
    /// Returns `0` when showing live output (bottom of buffer).
    /// Returns [`scrollback_available()`](Self::scrollback_available) when fully
    /// scrolled to the oldest history (top of buffer).
    #[must_use]
    pub fn scrollback_offset(&self) -> usize {
        self.parser_read().screen().scrollback()
    }

    /// Number of scrollback rows currently held in the buffer.
    ///
    /// Pair with [`scrollback_offset()`](Self::scrollback_offset) to drive a scrollbar:
    /// the offset ranges from `0` to this value.
    #[must_use]
    pub fn scrollback_available(&self) -> usize {
        self.parser_read().screen().scrollback_available()
    }

    /// Returns true if any mouse reporting mode is active.
    #[must_use]
    pub fn any_mouse_mode(&self) -> bool {
        let parser = self.parser_read();
        let s = parser.screen();
        s.mode(TerminalMode::MouseReportClick)
            || s.mode(TerminalMode::MouseReportCellMotion)
            || s.mode(TerminalMode::MouseReportAllMotion)
    }

    /// Returns the current Kitty keyboard enhancement flags (0 = none).
    #[must_use]
    pub fn kitty_keyboard_flags(&self) -> u8 {
        self.parser_read().screen().kitty_keyboard_flags()
    }
}

impl<M: TerminalOwnership> fmt::Debug for Terminal<M> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        let mut s = f.debug_struct("Terminal");
        if let Some(ref managed) = self.managed {
            s.field("child_pid", &managed.child_pid);
            s.field("finished", &poll_exit(managed).is_some());
        }
        s.finish_non_exhaustive()
    }
}

/// Dropping a `Terminal<Managed>` blocks the calling thread for up to
/// ~200 ms while it terminates the child process group and joins the
/// reader and writer threads. The 200 ms figure is a worst-case ceiling;
/// the typical cost is much shorter when the child cooperates with
/// SIGTERM, since each grace window wakes on actual child exit rather
/// than running to the full deadline.
///
/// * Send SIGTERM to the child process group.
/// * Wait for graceful exit, bounded by a 100 ms deadline. Returns as
///   soon as the child exits; the full 100 ms is paid only when the
///   child traps SIGTERM and runs cleanup up to the deadline.
/// * Send SIGKILL to the process group if the child is still alive.
/// * Join the reader and writer threads with a 50 ms timeout each.
///
/// If the child has already exited (e.g. you called [`Terminal::kill`] or
/// [`Terminal::wait_async`] returned), the SIGTERM path is skipped and only the
/// thread joins remain (~100 ms). `Terminal<Unmanaged>` skips the
/// child-termination phase entirely and only pays the join cost.
///
/// On a hot path (UI render loop, request handler), prefer one of:
///
/// * Call [`Terminal::kill`] before dropping the handle, so the SIGTERM
///   grace window overlaps with the work that follows.
/// * Perform the handle swap on a background thread (for example
///   `tokio::task::spawn_blocking(move || drop(old_session))`) so the
///   foreground task does not stall.
///
/// Drop is best-effort: if the SIGKILL escalation or thread joins fail,
/// the failure is logged but not surfaced. Consumers that require an
/// observable shutdown result should call [`Terminal::kill`] and inspect
/// its [`Result`] before dropping.
impl<M: TerminalOwnership> Drop for Terminal<M> {
    fn drop(&mut self) {
        if let Some(ref managed) = self.managed {
            let already_dead = poll_exit(managed).is_some();
            if !already_dead {
                if let Some(pid) = managed.child_pid {
                    let mut killer = managed.killer.lock().unwrap_or_else(|e| e.into_inner());
                    if let Err(err) = process_group::sigterm_group(pid, killer.as_mut()) {
                        warn!(
                            child_pid = pid,
                            error = %err,
                            "failed to send graceful termination to child on drop",
                        );
                    }
                }
                self.writer_tx.take();
                wait_for_exit_or_deadline(managed, SHUTDOWN_TIMEOUT);
                if poll_exit(managed).is_none()
                    && let Some(pid) = managed.child_pid
                {
                    let mut killer = managed.killer.lock().unwrap_or_else(|e| e.into_inner());
                    if let Err(err) = process_group::sigkill_group(pid, killer.as_mut()) {
                        warn!(
                            child_pid = pid,
                            error = %err,
                            "failed to force-kill child on drop",
                        );
                    }
                }
            } else {
                self.writer_tx.take();
            }
        } else {
            self.writer_tx.take();
        }
        join_with_timeout(self.reader_handle.take(), JOIN_TIMEOUT);
        join_with_timeout(self.writer_handle.take(), JOIN_TIMEOUT);
    }
}

const _: () = {
    fn _assert_send<T: Send>() {}
    fn _assert() {
        _assert_send::<Terminal<Managed>>();
        _assert_send::<Terminal<Unmanaged>>();
    }
};

#[cfg(test)]
mod tests {
    use super::*;
    use std::io::Cursor;
    use std::time::Duration;

    /// Poison the parser lock via a panicking writer on another thread.
    /// `parser_read()` / `parser_write()` use `PoisonError::into_inner`, so
    /// subsequent public accessors must keep working instead of propagating
    /// the poison as a panic. Only writer panics poison an RwLock (reader
    /// panics do not), so this is the only path that actually exercises the
    /// recovery code.
    #[test]
    fn parser_lock_recovers_from_poison() {
        let session: Terminal<Unmanaged> = Terminal::from_reader_writer(
            Box::new(Cursor::new(Vec::<u8>::new())),
            Box::new(std::io::sink()),
            SessionOptions::default(),
        )
        .expect("offline session build");

        let parser = Arc::clone(&session.parser);
        drop(
            std::thread::spawn(move || {
                let _guard = parser.write().unwrap();
                panic!("intentional poison");
            })
            .join(),
        );

        assert!(
            session.parser.is_poisoned(),
            "writer panic must poison lock"
        );

        let _: () = session.with_screen(|_| ());
        session.scroll_reset();
        assert!(
            session.parser.is_poisoned(),
            "lock stays poisoned; recovery does not clear the flag",
        );
    }

    #[test]
    fn from_reader_writer_returns_thread_spawn_errors() {
        fail_next_thread_spawn(io::WRITER_THREAD_NAME);

        let Err(err) = Terminal::<Unmanaged>::from_reader_writer(
            Box::new(Cursor::new(Vec::<u8>::new())),
            Box::new(std::io::sink()),
            SessionOptions::default(),
        ) else {
            panic!("offline session build unexpectedly succeeded");
        };

        assert!(
            matches!(
                err,
                Error::ThreadSpawn {
                    name: io::WRITER_THREAD_NAME,
                    ..
                }
            ),
            "expected writer thread spawn failure, got {err:?}",
        );
    }

    #[test]
    fn send_blocking_rejects_inside_tokio_runtime() {
        let session: Terminal<Unmanaged> = Terminal::from_reader_writer(
            Box::new(Cursor::new(Vec::<u8>::new())),
            Box::new(std::io::sink()),
            SessionOptions::default(),
        )
        .expect("offline session build");

        let rt = tokio::runtime::Builder::new_current_thread()
            .build()
            .unwrap();
        let err = rt.block_on(async { session.send_paste("x") }).unwrap_err();
        assert!(
            matches!(err, Error::BlockingInsideAsync),
            "expected BlockingInsideAsync, got {err:?}",
        );
    }

    /// A `Read` impl that yields one chunk, then a non-retryable error, to
    /// drive the reader thread into the new error path without spinning a
    /// real PTY.
    struct FaultyReader {
        sent_chunk: bool,
    }

    impl std::io::Read for FaultyReader {
        fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
            if !self.sent_chunk {
                self.sent_chunk = true;
                let payload = b"hi";
                let n = payload.len().min(buf.len());
                buf[..n].copy_from_slice(&payload[..n]);
                return Ok(n);
            }
            Err(std::io::Error::from(std::io::ErrorKind::PermissionDenied))
        }
    }

    #[test]
    fn reader_thread_surfaces_non_retryable_io_errors() {
        let session: Terminal<Unmanaged> = Terminal::from_reader_writer(
            Box::new(FaultyReader { sent_chunk: false }),
            Box::new(std::io::sink()),
            SessionOptions::default(),
        )
        .expect("offline session build");

        let mut rx = session
            .take_io_error_receiver()
            .expect("I/O error receiver");

        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_time()
            .build()
            .unwrap();
        let received = rt
            .block_on(async { tokio::time::timeout(Duration::from_secs(1), rx.recv()).await })
            .expect("reader error did not arrive in time")
            .expect("I/O error channel closed without sending");

        let crate::IoError::Reader(received) = received else {
            panic!("expected reader I/O error, got {received:?}");
        };

        assert_eq!(
            received.kind,
            std::io::ErrorKind::PermissionDenied,
            "expected the originating io::ErrorKind to be exposed verbatim",
        );
        assert_eq!(received.source.kind(), std::io::ErrorKind::PermissionDenied);
        // Source chain stays intact through the `#[source]` attribute.
        let chained: &dyn std::error::Error = &received;
        assert!(chained.source().is_some(), "source chain must be reachable");
    }

    /// Clean EOF (`Ok(0)`) does NOT produce a reader error: embedders that
    /// already wait on child exit should not be woken twice.
    #[test]
    fn reader_thread_keeps_eof_silent() {
        let session: Terminal<Unmanaged> = Terminal::from_reader_writer(
            Box::new(Cursor::new(Vec::<u8>::new())),
            Box::new(std::io::sink()),
            SessionOptions::default(),
        )
        .expect("offline session build");

        // Give the reader thread a moment to observe Ok(0) and exit.
        std::thread::sleep(Duration::from_millis(50));
        assert!(
            session.try_recv_io_error().is_none(),
            "EOF must not surface as a reader error",
        );
    }
}