sail/

shell.rs

//! Interactive PTY bridge: the local terminal driven against a `pty` exec.
//!
//! The exec RPC, output ring, and resize/stdin calls live in [`crate::exec`];
//! this module owns the terminal-facing half: putting the local TTY in raw
//! mode, forwarding raw keystrokes (so the guest's line discipline turns
//! Ctrl-C/Ctrl-D into signals), pumping merged output, propagating SIGWINCH
//! as a resize, and restoring the terminal on exit.
//!
//! [`Sailbox::shell`](crate::Sailbox::shell) is the high-level entry; the CLI
//! drives [`run_interactive`] directly for its `--tty` flows. 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::{GrpcCode, 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>,
}

/// 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: GrpcCode::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).
    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();
        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,
                    ..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: GrpcCode::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::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::SailError;
    use crate::exec::{ExecProcess, 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));

        drive_input(&proc, &stop);

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

        let exit = block_on(proc.wait())?;
        Ok(exit.exit_code)
    }

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

    /// 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}");
    }
}