sail-rs 0.4.0

//! Interactive terminal session against a `pty` command in a Sailbox.
//!
//! [`Sailbox::shell`](crate::Sailbox::shell) is the high-level entry; the CLI
//! drives [`run_interactive`] directly for its `--tty` flows. It puts the local
//! terminal in raw mode, forwards keystrokes (so Ctrl-C/Ctrl-D reach the remote
//! process as signals), renders the merged output, propagates window resizes,
//! and restores the terminal on exit. Unix-only: on other platforms the calls
//! return an unsupported error and the build still succeeds.

use std::sync::Arc;
use std::time::Duration;

use crate::error::{RpcStatus, SailError};
use crate::exec::ExecOptions;
use crate::sailbox::object::Sailbox;

/// Options for [`Sailbox::shell`].
#[derive(Debug, Clone, Default)]
pub struct ShellOptions {
    /// Login shell to run when no command is given (default: the guest's
    /// `$SHELL`, else `/bin/bash`). Ignored when a command is given.
    pub shell: Option<String>,
    /// `$TERM` for the remote pty (default: the local `$TERM`).
    pub term: Option<String>,
    /// Working directory for the session.
    pub cwd: Option<String>,
    /// Wall-clock limit for the session; `None` means no limit.
    pub timeout: Option<Duration>,
    /// Do not let the box open the user's browser or forward its localhost
    /// servers (both on by default). Set for an untrusted or automated session.
    pub no_forward: bool,
    /// Turn off browser opens only, keeping localhost server forwarding on.
    /// Ignored when `no_forward` is set.
    pub no_forward_browser: bool,
}

/// True when stdin and stdout are both TTYs, required for an interactive PTY.
#[doc(hidden)]
pub fn stdio_is_tty() -> bool {
    use std::io::IsTerminal;
    std::io::stdin().is_terminal() && std::io::stdout().is_terminal()
}

fn tty_required() -> SailError {
    SailError::Execution {
        code: RpcStatus::FailedPrecondition,
        detail: "shell requires an interactive terminal (stdin and stdout must be TTYs)"
            .to_string(),
    }
}

impl Sailbox {
    /// Open an interactive pty session on the Sailbox, driving the local
    /// terminal. With no `command`, runs a login shell; pass a command to run
    /// that under a pty instead (e.g. a REPL or an editor). Raw-mode
    /// keystrokes (including Ctrl-C, Ctrl-Z, and Ctrl-D) reach the remote
    /// process, its output renders locally, and terminal resizes propagate.
    /// Blocks until the remote process exits and returns its exit code.
    /// Requires an interactive local terminal (stdin and stdout TTYs). While
    /// the session is open, browser opens and localhost servers in the box
    /// are forwarded to the local machine; see [`ShellOptions::no_forward`].
    ///
    /// Runs on the local machine, which must be Unix (it needs Unix TTY and
    /// signal APIs).
    ///
    /// This is the one process-global API in the crate: for the session's
    /// duration it owns stdin/stdout, switches the terminal to raw mode, and
    /// installs a signal handler, restoring them when the session ends. The
    /// bridge runs on a blocking thread, so cancelling this future does not
    /// end the session; stop it by exiting the remote process.
    pub async fn shell(
        &self,
        command: Option<&str>,
        options: ShellOptions,
    ) -> Result<i32, SailError> {
        if !stdio_is_tty() {
            return Err(tty_required());
        }
        let command = match command {
            Some(command) => command.to_string(),
            None => login_shell_command(options.shell.as_deref()),
        };
        let (cols, rows) = terminal_size();
        // An interactive shell forwards the session's localhost servers and browser
        // opens to the user's machine unless opted out.
        let (forward_ports, forward_browser) =
            crate::exec::forward_flags(options.no_forward, options.no_forward_browser);
        let proc = self
            .client()
            .exec_shell(
                self.sailbox_id(),
                &command,
                ExecOptions {
                    timeout: options.timeout,
                    pty: true,
                    term: options
                        .term
                        .or_else(|| std::env::var("TERM").ok())
                        .unwrap_or_default(),
                    cols,
                    rows,
                    cwd: options.cwd,
                    forward_ports,
                    forward_browser,
                    ..Default::default()
                },
            )
            .await?;
        let proc = Arc::new(proc);
        tokio::task::spawn_blocking(move || run_interactive(proc))
            .await
            .map_err(|err| SailError::Internal {
                message: format!("shell bridge task failed: {err}"),
            })?
    }
}

/// The command for an interactive login session: `exec` the login shell so
/// `$0` and login semantics match ssh. An explicit shell is quoted so a path
/// with spaces runs as a literal program; the default stays unquoted so the
/// guest shell expands `$SHELL`.
fn login_shell_command(shell: Option<&str>) -> String {
    match shell {
        Some(shell) => format!("exec {} -l", crate::exec::sh_quote(shell)),
        None => "exec ${SHELL:-/bin/bash} -l".to_string(),
    }
}

/// The local terminal size as (cols, rows), defaulting to 80x24.
#[cfg(unix)]
#[doc(hidden)]
pub fn terminal_size() -> (u32, u32) {
    let mut size = libc::winsize {
        ws_row: 0,
        ws_col: 0,
        ws_xpixel: 0,
        ws_ypixel: 0,
    };
    let ok = unsafe { libc::ioctl(libc::STDOUT_FILENO, libc::TIOCGWINSZ, &raw mut size) } == 0;
    if ok && size.ws_col > 0 && size.ws_row > 0 {
        (u32::from(size.ws_col), u32::from(size.ws_row))
    } else {
        (80, 24)
    }
}

/// The local terminal size as (cols, rows), defaulting to 80x24.
#[cfg(not(unix))]
#[doc(hidden)]
pub fn terminal_size() -> (u32, u32) {
    (80, 24)
}

/// Interactive PTY sessions need Unix TTY and signal APIs.
#[cfg(not(unix))]
#[doc(hidden)]
pub fn run_interactive(_proc: Arc<crate::exec::ExecProcess>) -> Result<i32, SailError> {
    Err(SailError::Execution {
        code: RpcStatus::Unimplemented,
        detail: "interactive PTY sessions are not supported on this platform".to_string(),
    })
}

#[cfg(unix)]
#[doc(hidden)]
pub use unix::run_interactive;

#[cfg(unix)]
#[doc(hidden)]
pub use unix::drive_output_pump;

#[cfg(unix)]
mod unix {
    use std::collections::{HashMap, HashSet};
    use std::io::Write;
    use std::sync::atomic::{AtomicBool, Ordering};
    use std::sync::Arc;
    use std::thread;
    use std::time::{Duration, Instant};

    use super::terminal_size;
    use crate::error::{RpcStatus, SailError};
    use crate::exec::{ExecProcess, ForwardEvent, OutputStream, ReadStep};

    /// Drive a future to completion from this bridge's dedicated thread. On
    /// the shared runtime's blocking pool (the [`Sailbox::shell`] path) an
    /// ambient handle exists and `Handle::block_on` is the correct, safe
    /// call; on a plain thread (the CLI's direct `run_interactive` use) fall
    /// back to the crate's shared-runtime `block_on`.
    fn block_on<F: std::future::Future>(future: F) -> F::Output {
        match tokio::runtime::Handle::try_current() {
            Ok(handle) => handle.block_on(future),
            Err(_) => crate::runtime::block_on(future),
        }
    }

    /// Set by the SIGWINCH handler; drained by the input loop to issue a resize.
    static RESIZE_PENDING: AtomicBool = AtomicBool::new(false);

    extern "C" fn on_sigwinch(_signum: libc::c_int) {
        RESIZE_PENDING.store(true, Ordering::Relaxed);
    }

    /// Drive the local terminal against a PTY exec until the remote process
    /// exits, returning its exit code. Raw mode and the SIGWINCH handler are
    /// always restored, even on error.
    pub fn run_interactive(proc: Arc<ExecProcess>) -> Result<i32, SailError> {
        let saved = enter_raw_mode()?;
        let prev_winch = install_sigwinch();
        let prev_in_flags = set_stdin_nonblocking();
        // Non-blocking stdout so the output pump is never parked in a write to a
        // slow terminal: it must stay free to notice the ring dropped and repaint.
        let prev_out_flags = set_stdout_nonblocking();

        // Seed the remote PTY with the current size.
        let (cols, rows) = terminal_size();
        block_on(proc.resize(cols, rows));

        let stop = Arc::new(AtomicBool::new(false));
        let output = spawn_output_pump(Arc::clone(&proc), Arc::clone(&stop));
        let forward = spawn_forward_consumer(Arc::clone(&proc));

        drive_input(&proc, &stop);

        // Tear down in reverse order so the terminal is always usable afterwards.
        let _ = output.join();
        let _ = forward.join();
        restore_stdout_flags(prev_out_flags);
        restore_stdin_flags(prev_in_flags);
        restore_sigwinch(prev_winch);
        restore_terminal(&saved);

        // A witnessed Exit is the command's real result. When the stream ended
        // without one, the command did not exit; the box was parked (put to
        // sleep) or otherwise became unreachable mid-session. Report that instead
        // of calling wait(), which would block forever on an Exit an interactive
        // shell never emits; the box's session stays intact for a fresh reconnect.
        match proc.try_wait() {
            Some(result) => result,
            None => Err(SailError::Execution {
                code: RpcStatus::Unavailable,
                detail: format!(
                    "the box became unavailable and the shell session ended; \
                     reconnect with `sail box shell {}`",
                    proc.sailbox_id(),
                ),
            }),
        }
    }

    /// Least time between screen-repaint requests while the local terminal is
    /// too slow to keep up: without a bound a persistently-behind reader would
    /// ask on every drop and flood the guest with resync RPCs. Capping repaint
    /// requests to one per 100 ms is plenty to keep the screen current.
    const RESYNC_MIN_INTERVAL: Duration = Duration::from_millis(100);

    /// Most backlog the pump buffers toward the terminal before it stops draining
    /// the ring. Holding the cap small means a slow terminal quickly lets the
    /// ring back up and drop-oldest, which the reader reports as a drop — the
    /// signal that triggers a repaint. Larger would just make the terminal crawl
    /// further through stale frames before recovering.
    const OUTPUT_PENDING_CAP: usize = 256 * 1024;

    /// Spawn the thread that renders merged PTY output to the terminal, then
    /// signals stop when the stream ends. The terminal fd is already non-blocking
    /// (set by [`run_interactive`]).
    fn spawn_output_pump(proc: Arc<ExecProcess>, stop: Arc<AtomicBool>) -> thread::JoinHandle<()> {
        thread::spawn(move || {
            let mut reader = proc.reader(OutputStream::Stdout);
            let mut sink = RawFdWriter(libc::STDOUT_FILENO);
            drive_output_pump(&mut reader, &mut sink, &proc);
            stop.store(true, Ordering::Relaxed);
        })
    }

    /// Least time between retries of a port that could not be forwarded because
    /// its local port was busy. The port watcher only re-reports the guest's
    /// listeners when the set changes, so this retry covers a local port freeing
    /// up while the guest server keeps running.
    const FORWARD_RETRY_INTERVAL: Duration = Duration::from_secs(3);

    /// Spawn the thread that acts on the session's local-forwarding events. It
    /// drains until the stream ends (the accessor then returns `None`). Active
    /// port forwards are held for the life of the session and dropped on exit.
    fn spawn_forward_consumer(proc: Arc<ExecProcess>) -> thread::JoinHandle<()> {
        thread::spawn(move || {
            let mut forwards: HashMap<u16, crate::forward::PortForward> = HashMap::new();
            // Ports whose local port was busy, so the forward could not bind. Kept
            // so the bind is retried on the interval below in case the local port
            // frees up.
            let mut conflicts: HashSet<u16> = HashSet::new();
            loop {
                // Wait for the next event, but only until the retry interval when
                // there are conflicts to re-attempt; otherwise wait indefinitely.
                let next: Result<Option<ForwardEvent>, tokio::time::error::Elapsed> =
                    if conflicts.is_empty() {
                        Ok(block_on(proc.next_forward_event()))
                    } else {
                        block_on(async {
                            tokio::time::timeout(FORWARD_RETRY_INTERVAL, proc.next_forward_event())
                                .await
                        })
                    };
                match next {
                    Ok(Some(ForwardEvent::OpenUrl(url))) => {
                        let open = if !is_openable_scheme(&url) {
                            // open_local_url only opens http(s); skip building a
                            // forward for a URL it would refuse anyway.
                            false
                        } else if let Some(port) = crate::forward::forwardable_local_port(&url) {
                            // A URL for a server in the box: forward its port, then
                            // open the bound local address (rewritten below). If it
                            // can't be forwarded, don't open it against the user's
                            // own machine. A login's localhost callback is a server
                            // too, so the port watcher forwards it the same way.
                            if ensure_forward(&proc, &mut forwards, &mut conflicts, port) {
                                true
                            } else {
                                notify_local_port_busy(&url, port);
                                false
                            }
                        } else if crate::forward::is_unforwardable_loopback_url(&url) {
                            // A loopback URL the tunnel can't reach: opening it would
                            // hit the user's own machine, not the sandbox.
                            notify_loopback_unreachable(&url);
                            false
                        } else if let Some(callback) = crate::forward::redirect_callback(&url) {
                            // An external login URL whose redirect returns to a
                            // loopback callback. Don't start a login whose redirect,
                            // carrying the auth code, would hit the user's machine
                            // rather than the sandbox.
                            match callback {
                                // Open only once the callback port is actually
                                // forwarded. The snapshot precedes this URL, so a
                                // listening forwardable callback is already in
                                // `forwards`; a port not there is one whose local
                                // port is busy or whose server is not listening on a
                                // reachable address, and an immediate redirect would
                                // hit the user's own machine.
                                crate::forward::RedirectCallback::Forwardable(port)
                                    if forwards.contains_key(&port) =>
                                {
                                    true
                                }
                                crate::forward::RedirectCallback::Forwardable(port) => {
                                    notify_callback_unforwarded(&url, port);
                                    false
                                }
                                // A loopback the tunnel cannot dial at all.
                                crate::forward::RedirectCallback::Unreachable => {
                                    notify_callback_unreachable(&url);
                                    false
                                }
                            }
                        } else {
                            // An external URL with no localhost callback: open it.
                            true
                        };
                        if open {
                            let url = crate::forward::rewrite_loopback_url(&url, |remote| {
                                forwards
                                    .get(&remote)
                                    .map(crate::forward::PortForward::local_port)
                            });
                            open_local_url(&url);
                        }
                    }
                    Ok(Some(ForwardEvent::PortSnapshot(ports))) => {
                        // Reconcile against the authoritative set: drop forwards and
                        // conflicts for servers that are gone, then forward the rest.
                        let listening: HashSet<u16> = ports.iter().copied().collect();
                        forwards.retain(|port, _| listening.contains(port));
                        conflicts.retain(|port| listening.contains(port));
                        for port in ports {
                            ensure_forward(&proc, &mut forwards, &mut conflicts, port);
                        }
                    }
                    // The stream ended.
                    Ok(None) => break,
                    // No event within the interval: retry any port whose local port
                    // was busy, in case it has since freed up.
                    Err(_) => {
                        for port in conflicts.iter().copied().collect::<Vec<_>>() {
                            ensure_forward(&proc, &mut forwards, &mut conflicts, port);
                        }
                    }
                }
            }
        })
    }

    /// Forward `port` (guest to the same local port) if it is not already
    /// forwarded. Returns whether the port is now forwarded. The local port
    /// always matches the guest port and is never remapped: a login callback
    /// redirect targets that exact port, so binding elsewhere would send the
    /// browser to whatever already holds the local port rather than the sandbox.
    /// A busy local port is recorded in `conflicts` and retried on the interval.
    fn ensure_forward(
        proc: &Arc<ExecProcess>,
        forwards: &mut HashMap<u16, crate::forward::PortForward>,
        conflicts: &mut HashSet<u16>,
        port: u16,
    ) -> bool {
        if forwards.contains_key(&port) {
            return true;
        }
        if let Ok(forward) = block_on(proc.forward_port(port, port)) {
            forwards.insert(port, forward);
            conflicts.remove(&port);
            true
        } else {
            conflicts.insert(port);
            false
        }
    }

    /// Notify that a URL the box asked to open targets a loopback address the
    /// sandbox cannot reach, so it was not opened against the user's own machine.
    fn notify_loopback_unreachable(url: &str) {
        let notice = format!(
            "\r\n[sail] not opening {url}: it targets a loopback address the sandbox cannot reach\r\n"
        );
        let _ = RawFdWriter(libc::STDERR_FILENO).write_all(notice.as_bytes());
    }

    /// Notify that a box server was not opened because its port is already in use
    /// on the local machine, so the forward could not bind it.
    fn notify_local_port_busy(url: &str, port: u16) {
        let notice = format!("\r\n[sail] not opening {url}: local port {port} is in use\r\n");
        let _ = RawFdWriter(libc::STDERR_FILENO).write_all(notice.as_bytes());
    }

    /// Whether `open_local_url` would open this URL (it opens only http(s)).
    fn is_openable_scheme(url: &str) -> bool {
        url.starts_with("http://") || url.starts_with("https://")
    }

    /// Notify that a login was not opened because its localhost callback port is
    /// not forwarded (its local port is busy, or its server is not listening on a
    /// reachable address), so the provider's redirect could not reach the sandbox.
    fn notify_callback_unforwarded(url: &str, port: u16) {
        let notice = format!(
            "\r\n[sail] not opening {url}: its login callback port {port} is not forwarded\r\n"
        );
        let _ = RawFdWriter(libc::STDERR_FILENO).write_all(notice.as_bytes());
    }

    /// Notify that a login was not opened because its callback is a loopback
    /// address the sandbox cannot reach, so the redirect would hit the user's
    /// own machine.
    fn notify_callback_unreachable(url: &str) {
        let notice = format!(
            "\r\n[sail] not opening {url}: its login callback is a loopback address the sandbox cannot reach\r\n"
        );
        let _ = RawFdWriter(libc::STDERR_FILENO).write_all(notice.as_bytes());
    }

    /// Open a URL in the user's local browser, best-effort. Only http(s) URLs
    /// are opened, so a sandbox process cannot drive arbitrary local handlers.
    /// The child inherits no terminal, so an opener's own output can't corrupt
    /// the session. Silent on success: the browser tab appearing is the signal.
    fn open_local_url(url: &str) {
        if !is_openable_scheme(url) {
            return;
        }
        let _ = local_browser_command(url)
            .stdin(std::process::Stdio::null())
            .stdout(std::process::Stdio::null())
            .stderr(std::process::Stdio::null())
            .spawn();
    }

    /// The platform command that opens a URL in the default browser. This
    /// module is Unix-only, so the choice is macOS `open` or Linux `xdg-open`.
    fn local_browser_command(url: &str) -> std::process::Command {
        let program = if cfg!(target_os = "macos") {
            "open"
        } else {
            "xdg-open"
        };
        let mut command = std::process::Command::new(program);
        command.arg(url);
        command
    }

    /// Render one live output stream onto a terminal `sink` until the stream
    /// ends, favoring a current screen over a faithful replay.
    ///
    /// The terminal writer must never block the loop: a slow terminal has to keep
    /// the pump free to notice the ring dropped and ask the guest to repaint the
    /// current screen ([`ExecProcess::resync`]). So `sink` is written
    /// non-blockingly, backlog is held to [`OUTPUT_PENDING_CAP`] so the ring
    /// backs up and drops-oldest when the terminal falls behind, and a reported
    /// drop discards the torn backlog and requests a repaint rather than crawling
    /// the slow terminal through stale frames it will never catch. The command is
    /// detached on the server, so none of this ever blocks it.
    ///
    /// Generic over the sink so the drop-to-repaint behavior is testable against a
    /// deliberately slow writer without a real terminal.
    #[doc(hidden)]
    pub fn drive_output_pump<W: Write>(
        reader: &mut crate::exec::StreamReader,
        sink: &mut W,
        proc: &Arc<ExecProcess>,
    ) {
        let mut pending: Vec<u8> = Vec::new();
        let mut last_resync: Option<Instant> = None;
        // Hold an observed drop until a repaint is actually requested. resync_due
        // only fires once per RESYNC_MIN_INTERVAL, so a drop seen during that
        // cooldown would otherwise be forgotten, leaving the screen showing a
        // torn, partial frame.
        let mut resync_pending = false;
        loop {
            // Push as much backlog as the terminal accepts right now, without
            // blocking on it.
            let mut flushed = false;
            if !pending.is_empty() {
                let written = write_nonblocking(sink, &pending);
                if written > 0 {
                    pending.drain(..written);
                    flushed = true;
                }
            }
            // Refill from the ring, but only up to the cap: leaving the rest in
            // the ring lets it back up and drop-oldest when the terminal is slow.
            let mut progressed = false;
            if pending.len() < OUTPUT_PENDING_CAP {
                // Don't wait for new data while there is still backlog to push.
                let wait = if pending.is_empty() {
                    Duration::from_millis(50)
                } else {
                    Duration::ZERO
                };
                match reader.next(wait) {
                    ReadStep::Chunk(bytes) => {
                        // A Snapshot reset the ring: `bytes` is the repaint, and
                        // it supersedes the stale backlog buffered toward the
                        // terminal. Drop that backlog before queuing the repaint
                        // so the finished screen renders at once instead of stuck
                        // behind bytes the slow terminal will never finish
                        // draining (the bounded end-of-stream flush would give up
                        // before reaching it).
                        if reader.took_reset() {
                            pending.clear();
                        }
                        pending.extend_from_slice(&bytes);
                        progressed = true;
                    }
                    ReadStep::Eof => {
                        flush_blocking(sink, &pending);
                        return;
                    }
                    ReadStep::Pending => {}
                }
                while pending.len() < OUTPUT_PENDING_CAP {
                    match reader.try_next() {
                        // Honor a reset here too: the repaint can land in this
                        // batch drain when the Snapshot arrives after next()
                        // above already returned a stale chunk this iteration.
                        Some(more) => {
                            if reader.took_reset() {
                                pending.clear();
                            }
                            pending.extend_from_slice(&more);
                        }
                        None => break,
                    }
                }
            }
            // The ring evicted output we had not shown: the backlog is now a torn
            // tail, so drop it and repaint the current screen instead.
            if reader.took_drop() {
                pending.clear();
                resync_pending = true;
            }
            if resync_pending && resync_due(&mut last_resync) {
                resync_pending = false;
                let handle = Arc::clone(proc);
                crate::runtime::runtime().spawn(async move { handle.resync().await });
            }
            // Yield when no new ring data was read and bytes are still queued,
            // either because the backlog is at the cap (so the ring can back up
            // and drop-oldest for a slow terminal) or because the terminal is
            // back-pressured and accepted nothing (so the loop does not spin).
            // A terminal actively draining a partial backlog is making progress,
            // so it keeps looping.
            if !progressed
                && !pending.is_empty()
                && (pending.len() >= OUTPUT_PENDING_CAP || !flushed)
            {
                thread::sleep(Duration::from_millis(5));
            }
        }
    }

    /// Write what the terminal will take right now, returning the bytes accepted.
    /// A full terminal (`WouldBlock`), or any transient error, accepts zero and
    /// the caller keeps the rest rather than propagating a terminal write error.
    fn write_nonblocking<W: Write>(sink: &mut W, buf: &[u8]) -> usize {
        sink.write(buf).unwrap_or(0)
    }

    /// End of stream: land the final bytes even against a non-blocking terminal,
    /// but bounded so a wedged terminal cannot hang the exit.
    fn flush_blocking<W: Write>(sink: &mut W, buf: &[u8]) {
        let mut off = 0;
        for _ in 0..2000 {
            if off >= buf.len() {
                break;
            }
            match sink.write(&buf[off..]) {
                Ok(0) => thread::sleep(Duration::from_millis(1)),
                Ok(n) => off += n,
                Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => {
                    thread::sleep(Duration::from_millis(1));
                }
                Err(_) => break,
            }
        }
        let _ = sink.flush();
    }

    /// A `Write` over a raw fd. On a non-blocking fd a full pipe surfaces as a
    /// `WouldBlock` error rather than parking the thread.
    struct RawFdWriter(libc::c_int);

    impl Write for RawFdWriter {
        fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
            let n = unsafe { libc::write(self.0, buf.as_ptr().cast(), buf.len()) };
            if n < 0 {
                Err(std::io::Error::last_os_error())
            } else {
                Ok(n as usize)
            }
        }

        fn flush(&mut self) -> std::io::Result<()> {
            Ok(())
        }
    }

    /// Whether enough time has passed since the last repaint request to send
    /// another, stamping the clock when it returns true.
    fn resync_due(last: &mut Option<Instant>) -> bool {
        let now = Instant::now();
        if last.is_none_or(|t| now.duration_since(t) >= RESYNC_MIN_INTERVAL) {
            *last = Some(now);
            true
        } else {
            false
        }
    }

    /// Forward raw stdin bytes to the guest, draining pending resizes, until the
    /// output stream ends or local stdin closes.
    fn drive_input(proc: &Arc<ExecProcess>, stop: &AtomicBool) {
        let mut buf = [0u8; 4096];
        let mut stdin_open = true;
        while !stop.load(Ordering::Relaxed) {
            if RESIZE_PENDING.swap(false, Ordering::Relaxed) {
                let (cols, rows) = terminal_size();
                block_on(proc.resize(cols, rows));
            }
            if !stdin_open {
                thread::sleep(Duration::from_millis(20));
                continue;
            }
            let n = unsafe {
                libc::read(
                    libc::STDIN_FILENO,
                    buf.as_mut_ptr().cast::<libc::c_void>(),
                    buf.len(),
                )
            };
            match n.cmp(&0) {
                std::cmp::Ordering::Greater => {
                    if block_on(proc.write_stdin(&buf[..n as usize])).is_err() {
                        break; // remote closed stdin or exec ended
                    }
                }
                std::cmp::Ordering::Equal => {
                    // Local stdin reached EOF: send EOF and stop reading it, but
                    // keep draining output until the remote process exits.
                    let _ = block_on(proc.close_stdin());
                    stdin_open = false;
                }
                std::cmp::Ordering::Less => {
                    // A nonblocking read with no data yet (WouldBlock), or one a
                    // handled signal such as SIGWINCH interrupted (Interrupted),
                    // is transient: back off briefly and retry rather than ending
                    // the input loop, which would wedge stdin until the command
                    // exits.
                    let err = std::io::Error::last_os_error();
                    if matches!(
                        err.kind(),
                        std::io::ErrorKind::WouldBlock | std::io::ErrorKind::Interrupted
                    ) {
                        thread::sleep(Duration::from_millis(10));
                    } else {
                        break;
                    }
                }
            }
        }
    }

    // --- platform terminal plumbing ---

    /// Put the local terminal into raw mode, returning the saved settings.
    fn enter_raw_mode() -> Result<libc::termios, SailError> {
        unsafe {
            let mut saved: libc::termios = std::mem::zeroed();
            if libc::tcgetattr(libc::STDIN_FILENO, &raw mut saved) != 0 {
                return Err(SailError::Internal {
                    message: format!(
                        "could not enter raw terminal mode: {}",
                        std::io::Error::last_os_error()
                    ),
                });
            }
            let mut raw = saved;
            libc::cfmakeraw(&raw mut raw);
            if libc::tcsetattr(libc::STDIN_FILENO, libc::TCSADRAIN, &raw const raw) != 0 {
                return Err(SailError::Internal {
                    message: format!(
                        "could not enter raw terminal mode: {}",
                        std::io::Error::last_os_error()
                    ),
                });
            }
            Ok(saved)
        }
    }

    fn restore_terminal(saved: &libc::termios) {
        unsafe {
            let _ = libc::tcsetattr(
                libc::STDIN_FILENO,
                libc::TCSADRAIN,
                std::ptr::from_ref(saved),
            );
        }
    }

    type SigHandler = libc::sighandler_t;

    fn install_sigwinch() -> SigHandler {
        // `signal` takes the handler as a numeric `sighandler_t`; cast through a
        // concrete fn pointer first so this is a pointer-to-int cast, not a
        // fn-item-to-int cast.
        let handler = on_sigwinch as extern "C" fn(libc::c_int) as usize;
        unsafe { libc::signal(libc::SIGWINCH, handler) }
    }

    fn restore_sigwinch(prev: SigHandler) {
        unsafe {
            libc::signal(libc::SIGWINCH, prev);
        }
    }

    /// Put stdin into non-blocking mode so the input loop can interleave reads
    /// with resize handling and the stop flag. Returns the previous fcntl flags.
    fn set_stdin_nonblocking() -> libc::c_int {
        unsafe {
            let flags = libc::fcntl(libc::STDIN_FILENO, libc::F_GETFL);
            if flags >= 0 {
                libc::fcntl(libc::STDIN_FILENO, libc::F_SETFL, flags | libc::O_NONBLOCK);
            }
            flags
        }
    }

    fn restore_stdin_flags(flags: libc::c_int) {
        if flags >= 0 {
            unsafe {
                libc::fcntl(libc::STDIN_FILENO, libc::F_SETFL, flags);
            }
        }
    }

    /// Put stdout into non-blocking mode so the output pump is never parked in a
    /// write to a slow terminal. Returns the previous fcntl flags.
    fn set_stdout_nonblocking() -> libc::c_int {
        unsafe {
            let flags = libc::fcntl(libc::STDOUT_FILENO, libc::F_GETFL);
            if flags >= 0 {
                libc::fcntl(libc::STDOUT_FILENO, libc::F_SETFL, flags | libc::O_NONBLOCK);
            }
            flags
        }
    }

    fn restore_stdout_flags(flags: libc::c_int) {
        if flags >= 0 {
            unsafe {
                libc::fcntl(libc::STDOUT_FILENO, libc::F_SETFL, flags);
            }
        }
    }

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

        #[test]
        fn resync_due_throttles_back_to_back_requests() {
            let mut last = None;
            // The first request is always due and stamps the clock.
            assert!(resync_due(&mut last));
            // A second request within RESYNC_MIN_INTERVAL is suppressed, so a
            // persistently-behind reader cannot flood the guest with resync RPCs.
            assert!(!resync_due(&mut last));
        }
    }
}

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

    #[test]
    fn login_shell_quotes_an_explicit_path() {
        // A path with spaces runs as one literal program.
        assert_eq!(
            login_shell_command(Some("/opt/my tools/zsh")),
            "exec '/opt/my tools/zsh' -l"
        );
        // The default stays unquoted so the guest expands $SHELL.
        assert_eq!(
            login_shell_command(/* shell */ None),
            "exec ${SHELL:-/bin/bash} -l"
        );
    }

    #[tokio::test]
    async fn shell_requires_a_tty() {
        // Test processes have no TTY on stdin/stdout, so the precondition
        // fires before any network or terminal manipulation.
        let client = crate::Client::builder("sk_test")
            .api_url("http://127.0.0.1:1")
            .sailbox_api_url("http://127.0.0.1:1")
            .build()
            .expect("build");
        let err = client
            .sailbox("sb_test")
            .shell(/* command */ None, ShellOptions::default())
            .await
            .expect_err("no tty in tests");
        assert!(err.to_string().contains("interactive terminal"), "{err}");
    }
}