retach 0.10.0

use crate::pty::Pty;
use retach::screen::{Screen, TerminalSize};
use std::collections::{HashMap, VecDeque};
use std::io::{Read, Write};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};

/// Default terminal dimensions when actual size is unavailable.
pub const DEFAULT_COLS: u16 = 80;
pub const DEFAULT_ROWS: u16 = 24;

/// Maximum scrollback lines a client may request. The `history` field arrives
/// over the socket as an unbounded `usize` and is used directly as the
/// scrollback limit, so clamp it to prevent a memory-DoS via the socket.
pub const MAX_HISTORY: usize = 100_000;

/// RAII guard that clears `has_client` flag when dropped, unless evicted.
///
/// When a new client evicts the current one, it sends `false` on the eviction
/// channel. The guard checks this on Drop: if evicted, it skips clearing
/// has_client (the new client already set it to true).
pub struct ClientGuard {
    has_client: Arc<AtomicBool>,
    evict_rx: tokio::sync::watch::Receiver<bool>,
}

impl Drop for ClientGuard {
    fn drop(&mut self) {
        // evict_rx initial value is `true` (not evicted).
        // Eviction sends `false`.
        if *self.evict_rx.borrow() {
            self.has_client.store(false, Ordering::Release);
        }
    }
}

/// Session names are displayed in CLI output and stored as filesystem-friendly identifiers.
/// 128 bytes is generous for human-readable names while preventing abuse.
const MAX_SESSION_NAME_LEN: usize = 128;

/// PTY read buffer size. 4 KiB matches the typical pipe buffer granularity on
/// Linux/macOS and balances syscall overhead against latency.
const PTY_READ_BUF_SIZE: usize = 4096;

/// Maximum queued DA/DSR responses when pty_writer is contended. 64 is far
/// beyond normal usage (typically 1-2 responses in flight). Overflow means a
/// deadlock-like condition where the client holds pty_writer indefinitely.
const MAX_DEFERRED: usize = 64;

/// Shared screen handle.
pub type SharedScreen = Arc<Mutex<Screen>>;

/// Shared handles for the client relay tasks.
/// Created by `Session::connect()`.
#[derive(Clone)]
pub struct SessionHandles {
    pub screen: SharedScreen,
    pub pty_writer: crate::pty::SharedPtyWriter,
    pub master: crate::pty::SharedMasterPty,
    pub dims: Arc<Mutex<TerminalSize>>,
    pub screen_notify: Arc<tokio::sync::Notify>,
    pub reader_alive: Arc<AtomicBool>,
    /// Child exit status, populated by the persistent reader after PTY EOF.
    /// `None` while the child is running or when the code is unknown.
    pub exit_code: Arc<Mutex<Option<i32>>>,
    pub name: String,
}

/// A single terminal session backed by a PTY and a virtual screen.
pub struct Session {
    pub(crate) name: String,
    pub(crate) pty: Pty,
    pub(crate) screen: SharedScreen,
    pub(crate) dims: Arc<Mutex<TerminalSize>>,
    /// When a client is attached, holds the sender side of a watch channel.
    /// Sending `false` evicts the active client. Replaced on each new attach.
    evict_tx: Option<tokio::sync::watch::Sender<bool>>,
    /// Wakes the client relay when new PTY data has been processed.
    screen_notify: Arc<tokio::sync::Notify>,
    /// Whether a client is currently connected (used by reader to decide draining).
    has_client: Arc<AtomicBool>,
    /// Set to false when the persistent reader thread detects PTY EOF.
    reader_alive: Arc<AtomicBool>,
    /// Child exit status, populated by the persistent reader after PTY EOF.
    exit_code: Arc<Mutex<Option<i32>>>,
    /// Handle for the persistent PTY reader thread (joined on Drop).
    reader_handle: Option<std::thread::JoinHandle<()>>,
}

impl Session {
    /// Create a new session, spawning a shell in a PTY of the given size.
    pub fn new(name: String, cols: u16, rows: u16, history: usize) -> anyhow::Result<Self> {
        let pty = Pty::spawn(cols, rows)?;
        let screen = Arc::new(Mutex::new(Screen::new(cols, rows, history)));
        let dims = Arc::new(Mutex::new(TerminalSize { cols, rows }));
        let screen_notify = Arc::new(tokio::sync::Notify::new());
        let has_client = Arc::new(AtomicBool::new(false));
        let reader_alive = Arc::new(AtomicBool::new(true));
        let exit_code = Arc::new(Mutex::new(None));

        // Spawn the persistent PTY reader thread.
        let pty_reader = pty.clone_reader()?;
        let pty_writer = pty.writer_arc();
        let child = pty.child_arc();
        let reader_handle = {
            let screen = screen.clone();
            let notify = screen_notify.clone();
            let has_client = has_client.clone();
            let reader_alive = reader_alive.clone();
            let exit_code = exit_code.clone();
            let thread_name = format!("pty-reader-{}", name);
            std::thread::Builder::new()
                .name(thread_name)
                .spawn(move || {
                    persistent_reader_loop(
                        pty_reader,
                        screen,
                        pty_writer,
                        notify,
                        has_client,
                        reader_alive,
                        child,
                        exit_code,
                    );
                })?
        };

        Ok(Self {
            name,
            pty,
            screen,
            dims,
            evict_tx: None,
            screen_notify,
            has_client,
            reader_alive,
            exit_code,
            reader_handle: Some(reader_handle),
        })
    }

    /// Check if the session is still alive.
    /// A session is dead if the reader thread exited (PTY EOF or panic)
    /// or if the child process has terminated.
    pub fn is_alive(&self) -> bool {
        self.reader_alive.load(Ordering::Acquire) && self.pty.is_child_alive()
    }

    /// Get the child process PID (if available).
    /// Uses `try_lock()` to avoid blocking; returns `None` on contention or poison.
    pub fn child_pid(&self) -> Option<u32> {
        self.pty
            .child_arc()
            .try_lock()
            .ok()
            .and_then(|c| c.process_id())
    }

    /// Mark a client as connected, evicting any previous client.
    ///
    /// Returns a `ClientGuard` (clears `has_client` on drop), shared handles
    /// for the relay tasks, and an eviction watch receiver.
    ///
    /// **Must be called under the SessionManager lock** to prevent races.
    pub fn connect(
        &mut self,
    ) -> (
        ClientGuard,
        SessionHandles,
        tokio::sync::watch::Receiver<bool>,
    ) {
        // Set has_client BEFORE evicting old client, so the persistent reader
        // doesn't discard data intended for the new client.  Hold the screen
        // lock across the store so it is ordered against the reader's drain
        // critical section (which reads has_client while holding the same lock):
        // the reader either sees has_client=true and keeps the pending data, or
        // drains a batch that fully precedes this client's attach.
        {
            let _screen = self
                .screen
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner);
            self.has_client.store(true, Ordering::Release);
        }

        // Evict previous client if any
        if let Some(old_tx) = self.evict_tx.take() {
            tracing::debug!(session = %self.name, "evicting previous client");
            if old_tx.send(false).is_err() {
                tracing::debug!(session = %self.name, "evict channel: previous client already disconnected");
            }
        }

        // Create new eviction channel for this client
        let (evict_tx, evict_rx) = tokio::sync::watch::channel(true);
        self.evict_tx = Some(evict_tx);

        let guard = ClientGuard {
            has_client: self.has_client.clone(),
            evict_rx: evict_rx.clone(),
        };

        let handles = SessionHandles {
            screen: self.screen.clone(),
            pty_writer: self.pty.writer_arc(),
            master: self.pty.master_arc(),
            dims: self.dims.clone(),
            screen_notify: self.screen_notify.clone(),
            reader_alive: self.reader_alive.clone(),
            exit_code: self.exit_code.clone(),
            name: self.name.clone(),
        };

        (guard, handles, evict_rx)
    }

    /// Disconnect the current client (used by KillSession).
    /// Drops evict_tx so the connected client sees RecvError.
    pub fn disconnect(&mut self) {
        drop(self.evict_tx.take());
    }

    /// Test accessor: whether a client is connected.
    #[cfg(test)]
    pub(crate) fn has_client(&self) -> bool {
        self.has_client.load(Ordering::Acquire)
    }

    /// Test accessor: whether the reader thread is alive.
    #[cfg(test)]
    pub(crate) fn reader_alive(&self) -> bool {
        self.reader_alive.load(Ordering::Acquire)
    }

    /// Test accessor: the child's captured exit code (set after PTY EOF).
    #[cfg(test)]
    pub(crate) fn exit_code(&self) -> Option<i32> {
        self.exit_code.lock().ok().and_then(|c| *c)
    }
}

/// RAII guard that marks the reader dead and wakes the client on Drop.
///
/// Ensures that *any* exit from `persistent_reader_loop` — normal return, an
/// early `break`, or an unwinding panic inside `scr.process()` — stores
/// `reader_alive=false` and fires `screen_notify`. Without this, a panic would
/// leave `reader_alive=true` (session never reaped) and poison the screen mutex
/// (every reconnect failing forever).
struct ReaderDeathGuard {
    reader_alive: Arc<AtomicBool>,
    notify: Arc<tokio::sync::Notify>,
}

impl Drop for ReaderDeathGuard {
    fn drop(&mut self) {
        self.reader_alive.store(false, Ordering::Release);
        self.notify.notify_one();
    }
}

/// Persistent PTY reader loop, runs for the entire session lifetime.
/// Reads PTY output, feeds it through the screen's VTE parser, and notifies
/// any connected client of new data.
#[allow(clippy::too_many_arguments)]
fn persistent_reader_loop(
    mut reader: Box<dyn Read + Send>,
    screen: Arc<Mutex<Screen>>,
    pty_writer: Arc<Mutex<Box<dyn Write + Send>>>,
    notify: Arc<tokio::sync::Notify>,
    has_client: Arc<AtomicBool>,
    reader_alive: Arc<AtomicBool>,
    child: crate::pty::SharedChild,
    exit_code: Arc<Mutex<Option<i32>>>,
) {
    // On any exit (including panic), mark the reader dead and wake the client.
    let _death_guard = ReaderDeathGuard {
        reader_alive: reader_alive.clone(),
        notify: notify.clone(),
    };
    let mut buf = [0u8; PTY_READ_BUF_SIZE];
    let mut deferred_responses: VecDeque<Vec<u8>> = VecDeque::new();
    loop {
        match reader.read(&mut buf) {
            Ok(0) => {
                tracing::debug!("persistent pty reader: EOF");
                break;
            }
            Ok(n) => {
                let mut responses = {
                    let mut scr = match screen.lock() {
                        Ok(s) => s,
                        Err(e) => {
                            tracing::error!(error = %e, "screen mutex poisoned in reader loop, terminating");
                            break;
                        }
                    };
                    scr.process(&buf[..n]);
                    let responses = scr.take_responses();
                    // When no client is connected, drain pending data to prevent
                    // unbounded growth. The data is already in the main scrollback.
                    if !has_client.load(Ordering::Acquire) {
                        let _ = scr.take_pending_scrollback();
                        let _ = scr.take_passthrough();
                    }
                    responses
                };

                // Prepend any deferred responses from previous iterations.
                if !deferred_responses.is_empty() {
                    let mut all: Vec<Vec<u8>> = deferred_responses.drain(..).collect();
                    all.append(&mut responses);
                    responses = all;
                }

                // Write PTY responses (DA, DSR replies) outside the screen lock.
                // Use try_lock to avoid deadlock: client_to_pty may hold
                // pty_writer during a blocking write_all while waiting for the
                // child to read — if the child is waiting for this DA response,
                // a blocking lock() here would deadlock.
                if !responses.is_empty() {
                    match pty_writer.try_lock() {
                        Ok(mut w) => {
                            for response in &responses {
                                if let Err(e) = w.write_all(response) {
                                    tracing::warn!(error = %e, "failed to write response to PTY in reader loop");
                                    break;
                                }
                            }
                            if let Err(e) = w.flush() {
                                tracing::warn!(error = %e, "failed to flush PTY writer in reader loop");
                            }
                        }
                        Err(_) => {
                            tracing::debug!(
                                "pty_writer contended, deferring {} DA/DSR response(s)",
                                responses.len()
                            );
                            for resp in responses {
                                if deferred_responses.len() >= MAX_DEFERRED {
                                    tracing::warn!(queue_len = MAX_DEFERRED, "deferred DA/DSR response queue full, dropping oldest response");
                                    deferred_responses.pop_front();
                                }
                                deferred_responses.push_back(resp);
                            }
                        }
                    }
                }

                notify.notify_one();
            }
            Err(e) => {
                tracing::debug!(error = %e, "persistent pty reader: read error");
                break;
            }
        }
    }
    // Reap the child to capture its exit status so the client can propagate it
    // as its own exit code. PTY EOF means the child has exited, so wait() does
    // not block here. portable_pty maps a signal-terminated child to a non-zero
    // exit_code (1), so scripts checking $? still see failure.
    match child.lock() {
        Ok(mut c) => match c.wait() {
            Ok(status) => {
                if let Ok(mut slot) = exit_code.lock() {
                    *slot = Some(status.exit_code() as i32);
                }
            }
            Err(e) => {
                tracing::debug!(error = %e, "persistent pty reader: failed to wait for child");
            }
        },
        Err(e) => {
            tracing::warn!(error = %e, "child mutex poisoned in reader loop, cannot capture exit code");
        }
    }
    // `_death_guard` stores reader_alive=false and notifies on drop here.
}

impl Drop for Session {
    fn drop(&mut self) {
        // Use try_lock() to avoid blocking the async runtime if the child lock
        // is contended. In practice it's never contended during drop because
        // the persistent reader has already exited (or will exit after kill).
        match self.pty.child_arc().try_lock() {
            Ok(mut child) => {
                if let Err(e) = child.kill() {
                    tracing::debug!(error = %e, session = %self.name, "child already exited before kill");
                }
                if let Err(e) = child.wait() {
                    tracing::debug!(error = %e, session = %self.name, "child already reaped before wait");
                }
            }
            Err(_) => {
                tracing::warn!(session = %self.name, "child mutex contended during drop, skipping kill/wait — detaching reader thread");
                // Can't kill the child, so the PTY reader thread will block on read
                // forever. Detach it (take + drop the JoinHandle) instead of joining,
                // and skip eviction since the session is being abandoned.
                self.reader_handle.take();
                return;
            }
        }
        // Evict any connected client
        if let Some(tx) = self.evict_tx.take() {
            if tx.send(false).is_err() {
                tracing::debug!(session = %self.name, "evict channel: client already disconnected during drop");
            }
        }
        // Wait for the reader thread to exit (it will see EOF after child kill).
        if let Some(handle) = self.reader_handle.take() {
            if handle.join().is_err() {
                tracing::warn!(session = %self.name, "PTY reader thread panicked during join");
            }
        }
    }
}

/// Validate a session name: max 128 bytes, only `[a-zA-Z0-9_-.]`.
pub fn validate_session_name(name: &str) -> anyhow::Result<()> {
    if name.is_empty() {
        anyhow::bail!("session name cannot be empty");
    }
    if name.len() > MAX_SESSION_NAME_LEN {
        anyhow::bail!("session name too long (max {} bytes)", MAX_SESSION_NAME_LEN);
    }
    if let Some(ch) = name
        .chars()
        .find(|c| !matches!(c, 'a'..='z' | 'A'..='Z' | '0'..='9' | '_' | '-' | '.'))
    {
        anyhow::bail!(
            "invalid character '{}' in session name (allowed: a-zA-Z0-9_-.)",
            ch
        );
    }
    Ok(())
}

/// Registry of named sessions with create, lookup, and cleanup operations.
pub struct SessionManager {
    sessions: HashMap<String, Session>,
}

impl SessionManager {
    /// Create an empty session manager.
    pub fn new() -> Self {
        Self {
            sessions: HashMap::new(),
        }
    }

    /// Create a new session with the given name, failing if it already exists.
    /// Zero dimensions are clamped to defaults.
    pub fn create(
        &mut self,
        name: String,
        cols: u16,
        rows: u16,
        history: usize,
    ) -> anyhow::Result<()> {
        validate_session_name(&name)?;
        if self.sessions.contains_key(&name) {
            anyhow::bail!("session '{}' already exists", name);
        }
        let c = if cols > 0 { cols } else { DEFAULT_COLS };
        let r = if rows > 0 { rows } else { DEFAULT_ROWS };
        let session = Session::new(name.clone(), c, r, history.min(MAX_HISTORY))?;
        self.sessions.insert(name, session);
        Ok(())
    }

    /// Get existing session or create a new one.
    /// Returns (session, is_new).
    pub fn get_or_create(
        &mut self,
        name: &str,
        cols: u16,
        rows: u16,
        history: usize,
    ) -> anyhow::Result<(&mut Session, bool)> {
        validate_session_name(name)?;
        use std::collections::hash_map::Entry;
        match self.sessions.entry(name.to_string()) {
            Entry::Occupied(e) => {
                tracing::debug!(session = %name, "reattaching to existing session");
                Ok((e.into_mut(), false))
            }
            Entry::Vacant(e) => {
                let c = if cols > 0 { cols } else { DEFAULT_COLS };
                let r = if rows > 0 { rows } else { DEFAULT_ROWS };
                tracing::debug!(session = %name, cols = c, rows = r, "creating new session");
                let session = Session::new(name.to_string(), c, r, history.min(MAX_HISTORY))?;
                Ok((e.insert(session), true))
            }
        }
    }

    /// Get an existing session by name (read-only).
    pub fn get(&self, name: &str) -> Option<&Session> {
        self.sessions.get(name)
    }

    /// Get an existing session by name (mutable).
    pub fn get_mut(&mut self, name: &str) -> Option<&mut Session> {
        self.sessions.get_mut(name)
    }

    /// Remove and return a session by name, or `None` if not found.
    pub fn remove(&mut self, name: &str) -> Option<Session> {
        self.sessions.remove(name)
    }

    /// Return metadata for all active sessions.
    pub fn list(&self) -> Vec<crate::protocol::SessionInfo> {
        self.sessions.values().map(|s| {
            let dims = match s.dims.lock() {
                Ok(d) => *d,
                Err(e) => {
                    tracing::warn!(session = %s.name, error = %e, "dims mutex poisoned in list");
                    TerminalSize { cols: DEFAULT_COLS, rows: DEFAULT_ROWS }
                }
            };
            crate::protocol::SessionInfo {
                name: s.name.clone(),
                pid: s.child_pid().unwrap_or(0),
                cols: dims.cols,
                rows: dims.rows,
            }
        }).collect()
    }

    /// Remove and return all sessions (for graceful shutdown).
    pub fn drain_all(&mut self) -> Vec<Session> {
        self.sessions.drain().map(|(_, s)| s).collect()
    }

    /// Remove dead sessions and return them for cleanup outside the lock.
    pub fn take_dead_sessions(&mut self) -> Vec<Session> {
        let dead: Vec<String> = self
            .sessions
            .iter()
            .filter(|(_, s)| !s.is_alive())
            .map(|(name, s)| {
                // Use try_lock to avoid blocking a Tokio worker thread
                // (this is called from an async cleanup task).
                let status = s
                    .pty
                    .child_arc()
                    .try_lock()
                    .ok()
                    .and_then(|mut c| c.try_wait().ok().flatten());
                tracing::info!(
                    session = %name,
                    exit_status = ?status,
                    "cleaning up dead session"
                );
                name.clone()
            })
            .collect();
        dead.into_iter()
            .filter_map(|name| self.sessions.remove(&name))
            .collect()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::io::Write;
    use std::sync::atomic::Ordering;

    /// Helper: collect visible grid rows as trimmed strings.
    fn screen_lines(screen: &retach::screen::Screen) -> Vec<String> {
        screen
            .visible_rows()
            .map(|row| {
                let s: String = row.iter().map(|c| c.c).collect();
                s.trim_end().to_string()
            })
            .collect()
    }

    /// Helper: collect scrollback history as plain text (ANSI stripped).
    fn history_texts(screen: &retach::screen::Screen) -> Vec<String> {
        screen
            .get_history()
            .iter()
            .map(|b| {
                let s = String::from_utf8_lossy(b);
                let mut out = String::new();
                let mut in_esc = false;
                for ch in s.chars() {
                    if in_esc {
                        if ch.is_ascii_alphabetic() || ch == 'm' {
                            in_esc = false;
                        }
                        continue;
                    }
                    if ch == '\x1b' {
                        in_esc = true;
                        continue;
                    }
                    if ch >= ' ' {
                        out.push(ch);
                    }
                }
                out.trim_end().to_string()
            })
            .collect()
    }

    /// Poll the screen until a predicate is satisfied or timeout expires.
    fn wait_for_screen(
        screen: &Arc<Mutex<retach::screen::Screen>>,
        timeout: std::time::Duration,
        pred: impl Fn(&retach::screen::Screen) -> bool,
    ) -> bool {
        let start = std::time::Instant::now();
        while start.elapsed() < timeout {
            if let Ok(scr) = screen.lock() {
                if pred(&scr) {
                    return true;
                }
            }
            std::thread::sleep(std::time::Duration::from_millis(50));
        }
        false
    }

    /// Persistent reader processes PTY output while no client is connected.
    ///
    /// Simulates: client opens session running `sleep 2 && echo MARKER`,
    /// disconnects immediately, reconnects after the command completes,
    /// and finds MARKER in the screen or scrollback.
    #[test]
    fn persistent_reader_captures_output_without_client() {
        let session = Session::new("test-persistent".into(), 80, 24, 1000).unwrap();

        // No client connected — persistent reader is running with has_client=false.
        assert!(!session.has_client());
        assert!(session.reader_alive());

        // Write a command that produces output after a short delay.
        // Use a unique marker so we can find it unambiguously.
        {
            let writer = session.pty.writer_arc();
            let mut w = writer.lock().unwrap();
            w.write_all(b"sleep 1 && echo PERSISTENT_READER_OK\n")
                .unwrap();
            w.flush().unwrap();
        }

        // Wait for the marker to appear in the screen (up to 5s).
        let found = wait_for_screen(&session.screen, std::time::Duration::from_secs(5), |scr| {
            let lines = screen_lines(scr);
            let hist = history_texts(scr);
            lines
                .iter()
                .chain(hist.iter())
                .any(|l| l.contains("PERSISTENT_READER_OK"))
        });

        assert!(
            found,
            "persistent reader should capture PTY output even with no client connected"
        );

        // Reader should still be alive (shell is still running).
        assert!(session.reader_alive());
    }

    /// After the child process exits, a reconnecting client sees the final
    /// output and reader_alive is false.
    #[test]
    fn persistent_reader_detects_child_exit() {
        let session = Session::new("test-exit".into(), 80, 24, 1000).unwrap();

        // Tell the shell to print a marker and exit.
        {
            let writer = session.pty.writer_arc();
            let mut w = writer.lock().unwrap();
            w.write_all(b"echo GOODBYE && exit\n").unwrap();
            w.flush().unwrap();
        }

        // Wait for reader_alive to become false (child exited, PTY EOF).
        let exited = wait_for_screen(&session.screen, std::time::Duration::from_secs(5), |_| {
            !session.reader_alive()
        });
        assert!(exited, "reader_alive should become false after child exits");

        // The marker should be visible in the screen or scrollback.
        let scr = session.screen.lock().unwrap();
        let lines = screen_lines(&scr);
        let hist = history_texts(&scr);
        let found = lines
            .iter()
            .chain(hist.iter())
            .any(|l| l.contains("GOODBYE"));
        assert!(found, "final output should be captured before reader exits");
    }

    /// The persistent reader reaps the child after PTY EOF and records its
    /// non-zero exit code so the client can propagate it.
    #[test]
    fn persistent_reader_captures_exit_code() {
        let session = Session::new("test-exit-code".into(), 80, 24, 1000).unwrap();

        {
            let writer = session.pty.writer_arc();
            let mut w = writer.lock().unwrap();
            w.write_all(b"exit 7\n").unwrap();
            w.flush().unwrap();
        }

        // Wait for the reader to detect EOF and reap the child.
        let exited = wait_for_screen(&session.screen, std::time::Duration::from_secs(5), |_| {
            !session.reader_alive()
        });
        assert!(exited, "reader_alive should become false after child exits");

        assert_eq!(
            session.exit_code(),
            Some(7),
            "persistent reader should capture the child's exit code"
        );
    }

    #[test]
    fn create_clamps_huge_history() {
        // A client requesting an unbounded history must not OOM the daemon;
        // the clamp keeps allocation bounded. This succeeds quickly because the
        // scrollback grows lazily, so the request itself is the DoS surface.
        let mut mgr = SessionManager::new();
        assert!(mgr.create("clamp".into(), 80, 24, usize::MAX).is_ok());
        assert!(mgr.get_or_create("clamp2", 80, 24, usize::MAX).is_ok());
    }

    #[test]
    fn deferred_responses_bounded() {
        const { assert!(MAX_DEFERRED > 0 && MAX_DEFERRED <= 128) };
    }

    #[test]
    fn session_manager_create_and_list() {
        let mut mgr = SessionManager::new();
        mgr.create("test1".into(), 80, 24, 1000).unwrap();
        let list = mgr.list();
        assert_eq!(list.len(), 1);
        assert_eq!(list[0].name, "test1");
    }

    #[test]
    fn session_manager_duplicate_create_fails() {
        let mut mgr = SessionManager::new();
        mgr.create("test".into(), 80, 24, 1000).unwrap();
        assert!(mgr.create("test".into(), 80, 24, 1000).is_err());
    }

    #[test]
    fn session_manager_get_or_create() {
        let mut mgr = SessionManager::new();
        let (session, is_new) = mgr.get_or_create("test", 80, 24, 1000).unwrap();
        assert_eq!(session.name, "test");
        assert!(is_new);
        // Should return existing session
        let (session, is_new) = mgr.get_or_create("test", 80, 24, 1000).unwrap();
        assert_eq!(session.name, "test");
        assert!(!is_new);
        assert_eq!(mgr.list().len(), 1);
    }

    #[test]
    fn session_manager_remove() {
        let mut mgr = SessionManager::new();
        mgr.create("test".into(), 80, 24, 1000).unwrap();
        assert!(mgr.remove("test").is_some());
        assert!(mgr.remove("test").is_none());
        assert_eq!(mgr.list().len(), 0);
    }

    #[test]
    fn session_manager_get_or_create_zero_dimensions() {
        let mut mgr = SessionManager::new();
        let (session, is_new) = mgr.get_or_create("test", 0, 0, 1000).unwrap();
        // Should clamp to 80x24 defaults
        let dims = *session.dims.lock().unwrap();
        assert_eq!(dims.cols, 80);
        assert_eq!(dims.rows, 24);
        assert!(is_new);
    }

    #[test]
    fn validate_session_name_valid() {
        assert!(validate_session_name("my-session.1_OK").is_ok());
        assert!(validate_session_name("a").is_ok());
        assert!(validate_session_name(&"x".repeat(128)).is_ok());
    }

    #[test]
    fn validate_session_name_empty() {
        let err = validate_session_name("").unwrap_err();
        assert!(err.to_string().contains("empty"));
    }

    #[test]
    fn validate_session_name_too_long() {
        let err = validate_session_name(&"x".repeat(129)).unwrap_err();
        assert!(err.to_string().contains("too long"));
    }

    #[test]
    fn validate_session_name_invalid_chars() {
        assert!(validate_session_name("foo/bar").is_err());
        assert!(validate_session_name("foo bar").is_err());
        assert!(validate_session_name("foo\0bar").is_err());
        assert!(validate_session_name("../escape").is_err());
    }

    #[test]
    fn session_manager_rejects_invalid_names() {
        let mut mgr = SessionManager::new();
        assert!(mgr.create("bad/name".into(), 80, 24, 1000).is_err());
        assert!(mgr.get_or_create("bad name", 80, 24, 1000).is_err());
    }

    #[test]
    fn take_dead_sessions_returns_dead() {
        let mut mgr = SessionManager::new();
        mgr.create("alive".into(), 80, 24, 100).unwrap();
        mgr.create("doomed".into(), 80, 24, 100).unwrap();

        // Kill the doomed session's child process
        {
            let session = mgr.get_mut("doomed").unwrap();
            let child_arc = session.pty.child_arc();
            let mut child = child_arc.lock().unwrap();
            child.kill().ok();
            child.wait().ok();
        }

        // Poll take_dead_sessions until the reader thread detects EOF and the
        // doomed session is reaped. take_dead_sessions removes what it returns,
        // so accumulate across polls rather than relying on a single fixed wait.
        let mut dead: Vec<Session> = Vec::new();
        let start = std::time::Instant::now();
        while start.elapsed() < std::time::Duration::from_secs(5) {
            dead.extend(mgr.take_dead_sessions());
            if dead.iter().any(|s| s.name == "doomed") {
                break;
            }
            std::thread::sleep(std::time::Duration::from_millis(50));
        }

        let dead_names: Vec<&str> = dead.iter().map(|s| s.name.as_str()).collect();
        assert!(
            dead_names.contains(&"doomed"),
            "dead list should contain 'doomed': {:?}",
            dead_names
        );
        assert!(
            !dead_names.contains(&"alive"),
            "dead list should not contain 'alive': {:?}",
            dead_names
        );

        // Only 'alive' should remain
        assert_eq!(mgr.list().len(), 1);
        assert_eq!(mgr.list()[0].name, "alive");
    }

    #[test]
    fn take_dead_sessions_empty_when_all_alive() {
        let mut mgr = SessionManager::new();
        mgr.create("s1".into(), 80, 24, 100).unwrap();
        mgr.create("s2".into(), 80, 24, 100).unwrap();

        let dead = mgr.take_dead_sessions();
        assert!(
            dead.is_empty(),
            "no sessions should be dead: {:?}",
            dead.iter().map(|s| &s.name).collect::<Vec<_>>()
        );
        assert_eq!(mgr.list().len(), 2);
    }

    /// The reader death guard marks the reader dead and notifies on drop,
    /// so any exit (including a panic) leaves the session reapable.
    #[test]
    fn reader_death_guard_marks_dead_on_drop() {
        let reader_alive = Arc::new(AtomicBool::new(true));
        let notify = Arc::new(tokio::sync::Notify::new());
        {
            let _guard = ReaderDeathGuard {
                reader_alive: reader_alive.clone(),
                notify: notify.clone(),
            };
            assert!(reader_alive.load(Ordering::Acquire));
        }
        assert!(
            !reader_alive.load(Ordering::Acquire),
            "guard drop must clear reader_alive so the session is reaped"
        );
    }

    /// A panic while the guard is in scope still clears reader_alive — this is
    /// what protects against a poisoned screen mutex wedging the session.
    #[test]
    fn reader_death_guard_fires_on_panic() {
        let reader_alive = Arc::new(AtomicBool::new(true));
        let notify = Arc::new(tokio::sync::Notify::new());
        let ra = reader_alive.clone();
        let n = notify.clone();
        let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
            let _guard = ReaderDeathGuard {
                reader_alive: ra,
                notify: n,
            };
            panic!("simulated reader panic");
        }));
        assert!(result.is_err());
        assert!(
            !reader_alive.load(Ordering::Acquire),
            "guard must clear reader_alive even when unwinding from a panic"
        );
    }

    #[test]
    fn client_guard_clears_has_client_on_drop() {
        let has_client = Arc::new(AtomicBool::new(true));
        let (_evict_tx, evict_rx) = tokio::sync::watch::channel(true);
        {
            let _guard = ClientGuard {
                has_client: has_client.clone(),
                evict_rx,
            };
            assert!(has_client.load(Ordering::Acquire));
        }
        assert!(!has_client.load(Ordering::Acquire));
    }

    #[test]
    fn client_guard_skips_clear_when_evicted() {
        let has_client = Arc::new(AtomicBool::new(true));
        let (evict_tx, evict_rx) = tokio::sync::watch::channel(true);
        {
            let _guard = ClientGuard {
                has_client: has_client.clone(),
                evict_rx,
            };
            let _ = evict_tx.send(false);
        }
        assert!(has_client.load(Ordering::Acquire));
    }

    /// End-to-end eviction handshake: a second `connect()` evicts the first
    /// client. The first client's `evict_rx` observes the change to `false`,
    /// `has_client` REMAINS true after the first (evicted) guard drops — the
    /// documented race the connect() ordering protects against — and the second
    /// client's input still reaches the PTY.
    #[test]
    fn connect_eviction_handshake_keeps_has_client_for_new_client() {
        let mut session = Session::new("evict-handshake".into(), 80, 24, 1000).unwrap();

        // First client attaches.
        let (guard1, _handles1, mut evict_rx1) = session.connect();
        assert!(session.has_client(), "first connect sets has_client");
        assert!(
            *evict_rx1.borrow_and_update(),
            "first client starts un-evicted (watch value is true)"
        );

        // Second client attaches, evicting the first.
        let (_guard2, handles2, _evict_rx2) = session.connect();

        // The first client's receiver must observe the eviction. connect() sends
        // `false` on the old channel before dropping its sender, so the latest
        // observed value is `false` (the relay's `evict_rx.changed()` would have
        // resolved to Ok before the sender drop).
        assert!(
            !*evict_rx1.borrow_and_update(),
            "eviction sends `false` to the old client"
        );

        // Dropping the first (evicted) guard must NOT clear has_client: the new
        // client already set it. This is the race the connect() ordering guards.
        drop(guard1);
        assert!(
            session.has_client(),
            "evicted guard drop must not clear has_client out from under the new client"
        );

        // The second client's data flow is unaffected: input reaches the PTY and
        // the persistent reader echoes it back into the screen.
        {
            let mut w = handles2.pty_writer.lock().unwrap();
            w.write_all(b"echo SECOND_CLIENT_OK\n").unwrap();
            w.flush().unwrap();
        }
        let found = wait_for_screen(&session.screen, std::time::Duration::from_secs(5), |scr| {
            let lines = screen_lines(scr);
            let hist = history_texts(scr);
            lines
                .iter()
                .chain(hist.iter())
                .any(|l| l.contains("SECOND_CLIENT_OK"))
        });
        assert!(
            found,
            "second client's input must reach the PTY after eviction"
        );
    }
}