Skip to main content

sail/
shell.rs

1//! Interactive terminal session against a `pty` command in a Sailbox.
2//!
3//! [`Sailbox::shell`](crate::Sailbox::shell) is the high-level entry; the CLI
4//! drives [`run_interactive`] directly for its `--tty` flows. It puts the local
5//! terminal in raw mode, forwards keystrokes (so Ctrl-C/Ctrl-D reach the remote
6//! process as signals), renders the merged output, propagates window resizes,
7//! and restores the terminal on exit. Unix-only: on other platforms the calls
8//! return an unsupported error and the build still succeeds.
9
10use std::sync::Arc;
11use std::time::Duration;
12
13use crate::error::{RpcStatus, SailError};
14use crate::exec::ExecOptions;
15use crate::sailbox::object::Sailbox;
16
17/// Options for [`Sailbox::shell`].
18#[derive(Debug, Clone, Default)]
19pub struct ShellOptions {
20    /// Login shell to run when no command is given (default: the guest's
21    /// `$SHELL`, else `/bin/bash`). Ignored when a command is given.
22    pub shell: Option<String>,
23    /// `$TERM` for the remote pty (default: the local `$TERM`).
24    pub term: Option<String>,
25    /// Working directory for the session. `None` starts it in the image's
26    /// working directory, or `/` when the image does not set one.
27    pub cwd: Option<String>,
28    /// Wall-clock limit for the session; `None` means no limit.
29    pub timeout: Option<Duration>,
30    /// Turn off all local forwarding for the session (on by default): the
31    /// browser opens and localhost servers, plus paste, drag-and-drop, and
32    /// clipboard bridging. Every byte then passes through verbatim.
33    pub no_forward: bool,
34    /// Turn off forwarding the session's browser opens only, keeping everything
35    /// else forwarded. Ignored when `no_forward` is set.
36    pub no_forward_browser: bool,
37}
38
39/// True when stdin and stdout are both TTYs, required for an interactive PTY.
40#[doc(hidden)]
41pub fn stdio_is_tty() -> bool {
42    use std::io::IsTerminal;
43    std::io::stdin().is_terminal() && std::io::stdout().is_terminal()
44}
45
46fn tty_required() -> SailError {
47    SailError::Execution {
48        code: RpcStatus::FailedPrecondition,
49        detail: "shell requires an interactive terminal (stdin and stdout must be TTYs)"
50            .to_string(),
51    }
52}
53
54impl Sailbox {
55    /// Open an interactive pty session on the Sailbox, driving the local
56    /// terminal. With no `command`, runs a login shell; pass a command to run
57    /// that under a pty instead (e.g. a REPL or an editor). Raw-mode
58    /// keystrokes (including Ctrl-C, Ctrl-Z, and Ctrl-D) reach the remote
59    /// process, its output renders locally, and terminal resizes propagate.
60    /// Blocks until the remote process exits and returns its exit code.
61    /// Requires an interactive local terminal (stdin and stdout TTYs). While
62    /// the session is open, browser opens, localhost servers, paste, and
63    /// drag-and-drop are forwarded to the local machine, and Ctrl+V forwards
64    /// your clipboard (a two-way clipboard on devbox images, upload-and-paste
65    /// elsewhere); see [`ShellOptions::no_forward`].
66    ///
67    /// Runs on the local machine, which must be Unix (it needs Unix TTY and
68    /// signal APIs).
69    ///
70    /// This is the one process-global API in the crate: for the session's
71    /// duration it owns stdin/stdout, switches the terminal to raw mode, and
72    /// installs a signal handler, restoring them when the session ends. The
73    /// bridge runs on a blocking thread, so cancelling this future does not
74    /// end the session; stop it by exiting the remote process.
75    pub async fn shell(
76        &self,
77        command: Option<&str>,
78        options: ShellOptions,
79    ) -> Result<i32, SailError> {
80        if !stdio_is_tty() {
81            return Err(tty_required());
82        }
83        let command = match command {
84            Some(command) => command.to_string(),
85            None => login_shell_command(options.shell.as_deref()),
86        };
87        let (cols, rows) = terminal_size();
88        // An interactive shell forwards the session's localhost servers, browser
89        // opens, and clipboard/paste to the user's machine unless opted out.
90        let (forward_ports, forward_browser, forward_clipboard) =
91            crate::exec::forward_flags(options.no_forward, options.no_forward_browser);
92        let proc = self
93            .client()
94            .exec_shell(
95                self.sailbox_id(),
96                &command,
97                ExecOptions {
98                    timeout: options.timeout,
99                    pty: true,
100                    term: options
101                        .term
102                        .or_else(|| std::env::var("TERM").ok())
103                        .unwrap_or_default(),
104                    cols,
105                    rows,
106                    cwd: options.cwd,
107                    forward_ports,
108                    forward_browser,
109                    forward_clipboard,
110                    ..Default::default()
111                },
112            )
113            .await?;
114        let proc = Arc::new(proc);
115        tokio::task::spawn_blocking(move || run_interactive(proc))
116            .await
117            .map_err(|err| SailError::Internal {
118                message: format!("shell bridge task failed: {err}"),
119            })?
120    }
121}
122
123/// The command for an interactive login session: `exec` the login shell so
124/// `$0` and login semantics match ssh. An explicit shell is quoted so a path
125/// with spaces runs as a literal program; the default stays unquoted so the
126/// guest shell expands `$SHELL`.
127fn login_shell_command(shell: Option<&str>) -> String {
128    match shell {
129        Some(shell) => format!("exec {} -l", crate::exec::sh_quote(shell)),
130        None => "exec ${SHELL:-/bin/bash} -l".to_string(),
131    }
132}
133
134/// The local terminal size as (cols, rows), defaulting to 80x24.
135#[cfg(unix)]
136#[doc(hidden)]
137pub fn terminal_size() -> (u32, u32) {
138    let mut size = libc::winsize {
139        ws_row: 0,
140        ws_col: 0,
141        ws_xpixel: 0,
142        ws_ypixel: 0,
143    };
144    let ok = unsafe { libc::ioctl(libc::STDOUT_FILENO, libc::TIOCGWINSZ, &raw mut size) } == 0;
145    if ok && size.ws_col > 0 && size.ws_row > 0 {
146        (u32::from(size.ws_col), u32::from(size.ws_row))
147    } else {
148        (80, 24)
149    }
150}
151
152/// The local terminal size as (cols, rows), defaulting to 80x24.
153#[cfg(not(unix))]
154#[doc(hidden)]
155pub fn terminal_size() -> (u32, u32) {
156    (80, 24)
157}
158
159/// Interactive PTY sessions need Unix TTY and signal APIs.
160#[cfg(not(unix))]
161#[doc(hidden)]
162pub fn run_interactive(_proc: Arc<crate::exec::ExecProcess>) -> Result<i32, SailError> {
163    Err(SailError::Execution {
164        code: RpcStatus::Unimplemented,
165        detail: "interactive PTY sessions are not supported on this platform".to_string(),
166    })
167}
168
169#[cfg(unix)]
170#[doc(hidden)]
171pub use unix::run_interactive;
172
173#[cfg(unix)]
174#[doc(hidden)]
175pub use unix::{drive_output_pump, RenderControl};
176
177#[cfg(unix)]
178mod unix {
179    use std::collections::{HashMap, HashSet};
180    use std::io::Write;
181    use std::path::PathBuf;
182    use std::sync::atomic::{AtomicBool, Ordering};
183    use std::sync::Arc;
184    use std::thread;
185    use std::time::{Duration, Instant};
186
187    use super::terminal_size;
188    use crate::error::{RpcStatus, SailError};
189    use crate::exec::{ExecProcess, ForwardEvent, OutputStream, ReadStep};
190    use crate::shell_input::{
191        dropped_local_files, find, partial_suffix_len, sanitize_drop_name, InputEvent,
192        InputScanner, PASTE_END, PASTE_START,
193    };
194
195    /// Drive a future to completion from this bridge's dedicated thread. On
196    /// the shared runtime's blocking pool (the [`Sailbox::shell`] path) an
197    /// ambient handle exists and `Handle::block_on` is the correct, safe
198    /// call; on a plain thread (the CLI's direct `run_interactive` use) fall
199    /// back to the crate's shared-runtime `block_on`.
200    fn block_on<F: std::future::Future>(future: F) -> F::Output {
201        match tokio::runtime::Handle::try_current() {
202            Ok(handle) => handle.block_on(future),
203            Err(_) => crate::runtime::block_on(future),
204        }
205    }
206
207    /// Set by the SIGWINCH handler; drained by the input loop to issue a resize.
208    static RESIZE_PENDING: AtomicBool = AtomicBool::new(false);
209
210    extern "C" fn on_sigwinch(_signum: libc::c_int) {
211        RESIZE_PENDING.store(true, Ordering::Relaxed);
212    }
213
214    /// Drive the local terminal against a PTY exec until the remote process
215    /// exits, returning its exit code. Raw mode and the SIGWINCH handler are
216    /// always restored, even on error. When the session forwards its clipboard
217    /// (`proc.forward_clipboard()`), dragged files and Ctrl+V pastes forward
218    /// into the guest and in-guest copies mirror back to the local clipboard;
219    /// without it every byte passes through verbatim.
220    pub fn run_interactive(proc: Arc<ExecProcess>) -> Result<i32, SailError> {
221        let saved = enter_raw_mode()?;
222        let prev_winch = install_sigwinch();
223        let prev_in_flags = set_stdin_nonblocking();
224        // Non-blocking stdout so the output pump is never parked in a write to a
225        // slow terminal: it must stay free to notice the ring dropped and repaint.
226        let prev_out_flags = set_stdout_nonblocking();
227
228        // Seed the remote PTY with the current size.
229        let (cols, rows) = terminal_size();
230        block_on(proc.resize(cols, rows));
231
232        let stop = Arc::new(AtomicBool::new(false));
233        let render = Arc::new(RenderControl::default());
234        let output = spawn_output_pump(Arc::clone(&proc), Arc::clone(&stop), Arc::clone(&render));
235        // Browser-open and localhost-port forwarding (guest-gated by the launch
236        // flags); idles harmlessly when the session opted out.
237        let forward = spawn_forward_consumer(Arc::clone(&proc));
238        // The clipboard bridge, both directions, rides the same opt-in as the
239        // clipboard launch flag: mirror in-guest copies onto the local
240        // clipboard, and scan stdin for pastes/drags to send the other way.
241        let forward_clipboard = proc.forward_clipboard();
242        let clipboard = forward_clipboard.then(|| spawn_clipboard_consumer(Arc::clone(&proc)));
243
244        if forward_clipboard {
245            drive_input_forwarding(
246                PasteBridge::new(Arc::clone(&proc), Arc::clone(&render)),
247                &stop,
248            );
249        } else {
250            drive_input(&proc, &stop);
251        }
252
253        // Tear down in reverse order so the terminal is always usable afterwards.
254        let _ = output.join();
255        let _ = forward.join();
256        if let Some(consumer) = clipboard {
257            let _ = consumer.join();
258        }
259        restore_stdout_flags(prev_out_flags);
260        restore_stdin_flags(prev_in_flags);
261        restore_sigwinch(prev_winch);
262        restore_terminal(&saved);
263
264        // A witnessed Exit is the command's real result. When the stream ended
265        // without one, the command did not exit; the box was parked (put to
266        // sleep) or otherwise became unreachable mid-session. Report that instead
267        // of calling wait(), which would block forever on an Exit an interactive
268        // shell never emits; the box's session stays intact for a fresh reconnect.
269        match proc.try_wait() {
270            Some(result) => result,
271            None => Err(SailError::Execution {
272                code: RpcStatus::Unavailable,
273                detail: format!(
274                    "the box became unavailable and the shell session ended; \
275                     reconnect with `sail box shell {}`",
276                    proc.sailbox_id(),
277                ),
278            }),
279        }
280    }
281
282    /// Shared switches between the input side and the output pump: `paused`
283    /// stops the pump writing to the terminal while an upload progress line
284    /// owns it, and `bracketed_paste` tracks whether the guest application has
285    /// paste bracketing (DEC mode 2004) on, so injected pastes are framed the
286    /// way the terminal would frame a real one. Public only because the pump
287    /// is driven directly by integration tests.
288    #[doc(hidden)]
289    #[derive(Default)]
290    pub struct RenderControl {
291        paused: AtomicBool,
292        bracketed_paste: AtomicBool,
293    }
294
295    /// Least time between screen-repaint requests while the local terminal is
296    /// too slow to keep up: without a bound a persistently-behind reader would
297    /// ask on every drop and flood the guest with resync RPCs. Capping repaint
298    /// requests to one per 100 ms is plenty to keep the screen current.
299    const RESYNC_MIN_INTERVAL: Duration = Duration::from_millis(100);
300
301    /// Most backlog the pump buffers toward the terminal before it stops draining
302    /// the ring. Holding the cap small means a slow terminal quickly lets the
303    /// ring back up and drop-oldest, which the reader reports as a drop — the
304    /// signal that triggers a repaint. Larger would just make the terminal crawl
305    /// further through stale frames before recovering.
306    const OUTPUT_PENDING_CAP: usize = 256 * 1024;
307
308    /// Spawn the thread that renders merged PTY output to the terminal, then
309    /// signals stop when the stream ends. The terminal fd is already non-blocking
310    /// (set by [`run_interactive`]).
311    fn spawn_output_pump(
312        proc: Arc<ExecProcess>,
313        stop: Arc<AtomicBool>,
314        render: Arc<RenderControl>,
315    ) -> thread::JoinHandle<()> {
316        thread::spawn(move || {
317            let mut reader = proc.reader(OutputStream::Stdout);
318            let mut sink = RawFdWriter(libc::STDOUT_FILENO);
319            drive_output_pump(&mut reader, &mut sink, &proc, &render);
320            stop.store(true, Ordering::Relaxed);
321        })
322    }
323
324    /// Least time between retries of a port that could not be forwarded because
325    /// its local port was busy. The port watcher only re-reports the guest's
326    /// listeners when the set changes, so this retry covers a local port freeing
327    /// up while the guest server keeps running.
328    const FORWARD_RETRY_INTERVAL: Duration = Duration::from_secs(3);
329
330    /// Spawn the thread that acts on the session's local-forwarding events. It
331    /// drains until the stream ends (the accessor then returns `None`). Active
332    /// port forwards are held for the life of the session and dropped on exit.
333    fn spawn_forward_consumer(proc: Arc<ExecProcess>) -> thread::JoinHandle<()> {
334        thread::spawn(move || {
335            let mut forwards: HashMap<u16, crate::forward::PortForward> = HashMap::new();
336            // Ports whose local port was busy, so the forward could not bind. Kept
337            // so the bind is retried on the interval below in case the local port
338            // frees up.
339            let mut conflicts: HashSet<u16> = HashSet::new();
340            loop {
341                // Wait for the next event, but only until the retry interval when
342                // there are conflicts to re-attempt; otherwise wait indefinitely.
343                let next: Result<Option<ForwardEvent>, tokio::time::error::Elapsed> =
344                    if conflicts.is_empty() {
345                        Ok(block_on(proc.next_forward_event()))
346                    } else {
347                        block_on(async {
348                            tokio::time::timeout(FORWARD_RETRY_INTERVAL, proc.next_forward_event())
349                                .await
350                        })
351                    };
352                match next {
353                    Ok(Some(ForwardEvent::OpenUrl(url))) => {
354                        let open = if !is_openable_scheme(&url) {
355                            // open_local_url only opens http(s); skip building a
356                            // forward for a URL it would refuse anyway.
357                            false
358                        } else if let Some(port) = crate::forward::forwardable_local_port(&url) {
359                            // A URL for a server in the box: forward its port, then
360                            // open the bound local address (rewritten below). If it
361                            // can't be forwarded, don't open it against the user's
362                            // own machine. A login's localhost callback is a server
363                            // too, so the port watcher forwards it the same way.
364                            if ensure_forward(&proc, &mut forwards, &mut conflicts, port) {
365                                true
366                            } else {
367                                notify_local_port_busy(&url, port);
368                                false
369                            }
370                        } else if crate::forward::is_unforwardable_loopback_url(&url) {
371                            // A loopback URL the tunnel can't reach: opening it would
372                            // hit the user's own machine, not the sandbox.
373                            notify_loopback_unreachable(&url);
374                            false
375                        } else if let Some(callback) = crate::forward::redirect_callback(&url) {
376                            // An external login URL whose redirect returns to a
377                            // loopback callback. Don't start a login whose redirect,
378                            // carrying the auth code, would hit the user's machine
379                            // rather than the sandbox.
380                            match callback {
381                                // Open only once the callback port is actually
382                                // forwarded. The snapshot precedes this URL, so a
383                                // listening forwardable callback is already in
384                                // `forwards`; a port not there is one whose local
385                                // port is busy or whose server is not listening on a
386                                // reachable address, and an immediate redirect would
387                                // hit the user's own machine.
388                                crate::forward::RedirectCallback::Forwardable(port)
389                                    if forwards.contains_key(&port) =>
390                                {
391                                    true
392                                }
393                                crate::forward::RedirectCallback::Forwardable(port) => {
394                                    notify_callback_unforwarded(&url, port);
395                                    false
396                                }
397                                // A loopback the tunnel cannot dial at all.
398                                crate::forward::RedirectCallback::Unreachable => {
399                                    notify_callback_unreachable(&url);
400                                    false
401                                }
402                            }
403                        } else {
404                            // An external URL with no localhost callback: open it.
405                            true
406                        };
407                        if open {
408                            let url = crate::forward::rewrite_loopback_url(&url, |remote| {
409                                forwards
410                                    .get(&remote)
411                                    .map(crate::forward::PortForward::local_port)
412                            });
413                            open_local_url(&url);
414                        }
415                    }
416                    Ok(Some(ForwardEvent::PortSnapshot(ports))) => {
417                        // Reconcile against the authoritative set: drop forwards and
418                        // conflicts for servers that are gone, then forward the rest.
419                        let listening: HashSet<u16> = ports.iter().copied().collect();
420                        forwards.retain(|port, _| listening.contains(port));
421                        conflicts.retain(|port| listening.contains(port));
422                        for port in ports {
423                            ensure_forward(&proc, &mut forwards, &mut conflicts, port);
424                        }
425                    }
426                    // The stream ended.
427                    Ok(None) => break,
428                    // No event within the interval: retry any port whose local port
429                    // was busy, in case it has since freed up.
430                    Err(_) => {
431                        for port in conflicts.iter().copied().collect::<Vec<_>>() {
432                            ensure_forward(&proc, &mut forwards, &mut conflicts, port);
433                        }
434                    }
435                }
436            }
437        })
438    }
439
440    /// Forward `port` (guest to the same local port) if it is not already
441    /// forwarded. Returns whether the port is now forwarded. The local port
442    /// always matches the guest port and is never remapped: a login callback
443    /// redirect targets that exact port, so binding elsewhere would send the
444    /// browser to whatever already holds the local port rather than the sandbox.
445    /// A busy local port is recorded in `conflicts` and retried on the interval.
446    fn ensure_forward(
447        proc: &Arc<ExecProcess>,
448        forwards: &mut HashMap<u16, crate::forward::PortForward>,
449        conflicts: &mut HashSet<u16>,
450        port: u16,
451    ) -> bool {
452        if forwards.contains_key(&port) {
453            return true;
454        }
455        if let Ok(forward) = block_on(proc.forward_port(port, port)) {
456            forwards.insert(port, forward);
457            conflicts.remove(&port);
458            true
459        } else {
460            conflicts.insert(port);
461            false
462        }
463    }
464
465    /// Notify that a URL the Sailbox asked to open targets a loopback address the
466    /// sandbox cannot reach, so it was not opened against the user's own machine.
467    fn notify_loopback_unreachable(url: &str) {
468        let notice = format!(
469            "\r\n[sail] not opening {url}: it targets a loopback address the sandbox cannot reach\r\n"
470        );
471        let _ = RawFdWriter(libc::STDERR_FILENO).write_all(notice.as_bytes());
472    }
473
474    /// Notify that a Sailbox server was not opened because its port is already in use
475    /// on the local machine, so the forward could not bind it.
476    fn notify_local_port_busy(url: &str, port: u16) {
477        let notice = format!("\r\n[sail] not opening {url}: local port {port} is in use\r\n");
478        let _ = RawFdWriter(libc::STDERR_FILENO).write_all(notice.as_bytes());
479    }
480
481    /// Whether `open_local_url` would open this URL (it opens only http(s)).
482    fn is_openable_scheme(url: &str) -> bool {
483        url.starts_with("http://") || url.starts_with("https://")
484    }
485
486    /// Notify that a login was not opened because its localhost callback port is
487    /// not forwarded (its local port is busy, or its server is not listening on a
488    /// reachable address), so the provider's redirect could not reach the sandbox.
489    fn notify_callback_unforwarded(url: &str, port: u16) {
490        let notice = format!(
491            "\r\n[sail] not opening {url}: its login callback port {port} is not forwarded\r\n"
492        );
493        let _ = RawFdWriter(libc::STDERR_FILENO).write_all(notice.as_bytes());
494    }
495
496    /// Notify that a login was not opened because its callback is a loopback
497    /// address the sandbox cannot reach, so the redirect would hit the user's
498    /// own machine.
499    fn notify_callback_unreachable(url: &str) {
500        let notice = format!(
501            "\r\n[sail] not opening {url}: its login callback is a loopback address the sandbox cannot reach\r\n"
502        );
503        let _ = RawFdWriter(libc::STDERR_FILENO).write_all(notice.as_bytes());
504    }
505
506    /// Open a URL in the user's local browser, best-effort. Only http(s) URLs
507    /// are opened, so a sandbox process cannot drive arbitrary local handlers.
508    /// The child inherits no terminal, so an opener's own output can't corrupt
509    /// the session. Silent on success: the browser tab appearing is the signal.
510    fn open_local_url(url: &str) {
511        if !is_openable_scheme(url) {
512            return;
513        }
514        let _ = local_browser_command(url)
515            .stdin(std::process::Stdio::null())
516            .stdout(std::process::Stdio::null())
517            .stderr(std::process::Stdio::null())
518            .spawn();
519    }
520
521    /// The platform command that opens a URL in the default browser. This
522    /// module is Unix-only, so the choice is macOS `open` or Linux `xdg-open`.
523    fn local_browser_command(url: &str) -> std::process::Command {
524        let program = if cfg!(target_os = "macos") {
525            "open"
526        } else {
527            "xdg-open"
528        };
529        let mut command = std::process::Command::new(program);
530        command.arg(url);
531        command
532    }
533
534    /// Render one live output stream onto a terminal `sink` until the stream
535    /// ends, favoring a current screen over a faithful replay.
536    ///
537    /// The terminal writer must never block the loop: a slow terminal has to keep
538    /// the pump free to notice the ring dropped and ask the guest to repaint the
539    /// current screen ([`ExecProcess::resync`]). So `sink` is written
540    /// non-blockingly, backlog is held to [`OUTPUT_PENDING_CAP`] so the ring
541    /// backs up and drops-oldest when the terminal falls behind, and a reported
542    /// drop discards the torn backlog and requests a repaint rather than crawling
543    /// the slow terminal through stale frames it will never catch. The command is
544    /// detached on the server, so none of this ever blocks it.
545    ///
546    /// Generic over the sink so the drop-to-repaint behavior is testable against a
547    /// deliberately slow writer without a real terminal.
548    #[doc(hidden)]
549    pub fn drive_output_pump<W: Write>(
550        reader: &mut crate::exec::StreamReader,
551        sink: &mut W,
552        proc: &Arc<ExecProcess>,
553        render: &RenderControl,
554    ) {
555        let mut pending: Vec<u8> = Vec::new();
556        let mut last_resync: Option<Instant> = None;
557        // Hold an observed drop until a repaint is actually requested. resync_due
558        // only fires once per RESYNC_MIN_INTERVAL, so a drop seen during that
559        // cooldown would otherwise be forgotten, leaving the screen showing a
560        // torn, partial frame.
561        let mut resync_pending = false;
562        let mut modes = BracketedPasteTracker::default();
563        loop {
564            // An upload progress line owns the terminal: stop rendering (and
565            // stop draining the ring, which then backs up and drops-oldest just
566            // like a slow terminal — the existing repaint path heals it).
567            if render.paused.load(Ordering::Relaxed) {
568                thread::sleep(Duration::from_millis(5));
569                continue;
570            }
571            // Push as much backlog as the terminal accepts right now, without
572            // blocking on it.
573            let mut flushed = false;
574            if !pending.is_empty() {
575                let written = write_nonblocking(sink, &pending);
576                if written > 0 {
577                    pending.drain(..written);
578                    flushed = true;
579                }
580            }
581            // Refill from the ring, but only up to the cap: leaving the rest in
582            // the ring lets it back up and drop-oldest when the terminal is slow.
583            let mut progressed = false;
584            if pending.len() < OUTPUT_PENDING_CAP {
585                // Don't wait for new data while there is still backlog to push.
586                let wait = if pending.is_empty() {
587                    Duration::from_millis(50)
588                } else {
589                    Duration::ZERO
590                };
591                match reader.next(wait) {
592                    ReadStep::Chunk(bytes) => {
593                        // A Snapshot reset the ring: `bytes` is the repaint, and
594                        // it supersedes the stale backlog buffered toward the
595                        // terminal. Drop that backlog before queuing the repaint
596                        // so the finished screen renders at once instead of stuck
597                        // behind bytes the slow terminal will never finish
598                        // draining (the bounded end-of-stream flush would give up
599                        // before reaching it).
600                        if reader.took_reset() {
601                            pending.clear();
602                        }
603                        modes.scan(&bytes, render);
604                        pending.extend_from_slice(&bytes);
605                        progressed = true;
606                    }
607                    ReadStep::Eof => {
608                        flush_blocking(sink, &pending);
609                        return;
610                    }
611                    ReadStep::Pending => {}
612                }
613                while pending.len() < OUTPUT_PENDING_CAP {
614                    match reader.try_next() {
615                        // Honor a reset here too: the repaint can land in this
616                        // batch drain when the Snapshot arrives after next()
617                        // above already returned a stale chunk this iteration.
618                        Some(more) => {
619                            if reader.took_reset() {
620                                pending.clear();
621                            }
622                            modes.scan(&more, render);
623                            pending.extend_from_slice(&more);
624                        }
625                        None => break,
626                    }
627                }
628            }
629            // The ring evicted output we had not shown: the backlog is now a torn
630            // tail, so drop it and repaint the current screen instead.
631            if reader.took_drop() {
632                pending.clear();
633                resync_pending = true;
634            }
635            if resync_pending && resync_due(&mut last_resync) {
636                resync_pending = false;
637                let handle = Arc::clone(proc);
638                crate::runtime::runtime().spawn(async move { handle.resync().await });
639            }
640            // Yield when no new ring data was read and bytes are still queued,
641            // either because the backlog is at the cap (so the ring can back up
642            // and drop-oldest for a slow terminal) or because the terminal is
643            // back-pressured and accepted nothing (so the loop does not spin).
644            // A terminal actively draining a partial backlog is making progress,
645            // so it keeps looping.
646            if !progressed
647                && !pending.is_empty()
648                && (pending.len() >= OUTPUT_PENDING_CAP || !flushed)
649            {
650                thread::sleep(Duration::from_millis(5));
651            }
652        }
653    }
654
655    /// Write what the terminal will take right now, returning the bytes accepted.
656    /// A full terminal (`WouldBlock`), or any transient error, accepts zero and
657    /// the caller keeps the rest rather than propagating a terminal write error.
658    fn write_nonblocking<W: Write>(sink: &mut W, buf: &[u8]) -> usize {
659        sink.write(buf).unwrap_or(0)
660    }
661
662    /// End of stream: land the final bytes even against a non-blocking terminal,
663    /// but bounded so a wedged terminal cannot hang the exit.
664    fn flush_blocking<W: Write>(sink: &mut W, buf: &[u8]) {
665        let mut off = 0;
666        for _ in 0..2000 {
667            if off >= buf.len() {
668                break;
669            }
670            match sink.write(&buf[off..]) {
671                Ok(0) => thread::sleep(Duration::from_millis(1)),
672                Ok(n) => off += n,
673                Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => {
674                    thread::sleep(Duration::from_millis(1));
675                }
676                Err(_) => break,
677            }
678        }
679        let _ = sink.flush();
680    }
681
682    /// A `Write` over a raw fd. On a non-blocking fd a full pipe surfaces as a
683    /// `WouldBlock` error rather than parking the thread.
684    struct RawFdWriter(libc::c_int);
685
686    impl Write for RawFdWriter {
687        fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
688            let n = unsafe { libc::write(self.0, buf.as_ptr().cast(), buf.len()) };
689            if n < 0 {
690                Err(std::io::Error::last_os_error())
691            } else {
692                Ok(n as usize)
693            }
694        }
695
696        fn flush(&mut self) -> std::io::Result<()> {
697            Ok(())
698        }
699    }
700
701    /// Whether enough time has passed since the last repaint request to send
702    /// another, stamping the clock when it returns true.
703    fn resync_due(last: &mut Option<Instant>) -> bool {
704        let now = Instant::now();
705        if last.is_none_or(|t| now.duration_since(t) >= RESYNC_MIN_INTERVAL) {
706            *last = Some(now);
707            true
708        } else {
709            false
710        }
711    }
712
713    /// Forward raw stdin bytes to the guest, draining pending resizes, until the
714    /// output stream ends or local stdin closes.
715    fn drive_input(proc: &Arc<ExecProcess>, stop: &AtomicBool) {
716        let mut buf = [0u8; 4096];
717        let mut stdin_open = true;
718        while !stop.load(Ordering::Relaxed) {
719            if RESIZE_PENDING.swap(false, Ordering::Relaxed) {
720                let (cols, rows) = terminal_size();
721                block_on(proc.resize(cols, rows));
722            }
723            if !stdin_open {
724                thread::sleep(Duration::from_millis(20));
725                continue;
726            }
727            let n = unsafe {
728                libc::read(
729                    libc::STDIN_FILENO,
730                    buf.as_mut_ptr().cast::<libc::c_void>(),
731                    buf.len(),
732                )
733            };
734            match n.cmp(&0) {
735                std::cmp::Ordering::Greater => {
736                    if block_on(proc.write_stdin(&buf[..n as usize])).is_err() {
737                        break; // remote closed stdin or exec ended
738                    }
739                }
740                std::cmp::Ordering::Equal => {
741                    // Local stdin reached EOF: send EOF and stop reading it, but
742                    // keep draining output until the remote process exits.
743                    let _ = block_on(proc.close_stdin());
744                    stdin_open = false;
745                }
746                std::cmp::Ordering::Less => {
747                    // A nonblocking read with no data yet (WouldBlock), or one a
748                    // handled signal such as SIGWINCH interrupted (Interrupted),
749                    // is transient: back off briefly and retry rather than ending
750                    // the input loop, which would wedge stdin until the command
751                    // exits.
752                    let err = std::io::Error::last_os_error();
753                    if matches!(
754                        err.kind(),
755                        std::io::ErrorKind::WouldBlock | std::io::ErrorKind::Interrupted
756                    ) {
757                        thread::sleep(Duration::from_millis(10));
758                    } else {
759                        break;
760                    }
761                }
762            }
763        }
764    }
765
766    // --- local paste and drag-and-drop forwarding ---
767
768    /// Parent of the per-session directory where forwarded drops and pastes
769    /// land. Each session gets its own subdirectory (see PasteBridge::new) so
770    /// two shells' drops of the same name never collide, and a cancel/failure
771    /// rollback only ever deletes files this session uploaded.
772    const GUEST_DROPS_ROOT: &str = "/tmp/sail-drops";
773    /// Wait this long before drawing the upload progress line: most drops are
774    /// small images that land invisibly fast, and flashing a progress line for
775    /// them would just flicker the screen.
776    const UPLOAD_UI_DELAY: Duration = Duration::from_millis(400);
777    /// Largest content pushed onto the guest clipboard in one RPC (the message
778    /// must fit the transport's 4 MiB frame). Larger images upload as files
779    /// and paste as a guest path instead.
780    const CLIPBOARD_PUSH_MAX: usize = 3 * 1024 * 1024;
781    /// Longest a clipboard push may hold the input thread. The push must
782    /// complete before the Ctrl+V chord is forwarded (or the guest
783    /// application would paste the clipboard's previous content), and the
784    /// input thread is what sequences both, so a slow guest stalls keystroke
785    /// forwarding for the push's duration. This deadline bounds that stall
786    /// well under the RPC's own; on expiry the caller falls back exactly as
787    /// for any other failed push. It must exceed the guest agent's own 3s
788    /// serve-verification budget (saild's guestClipboardCmdTimeout) with
789    /// round-trip slack, or a push the guest completed near its budget would
790    /// be misclassified as failed.
791    const CLIPBOARD_PUSH_DEADLINE: Duration = Duration::from_secs(5);
792    /// Upload stream granularity: small enough for responsive progress, large
793    /// enough that per-message overhead is noise.
794    const UPLOAD_CHUNK_BYTES: usize = 256 * 1024;
795    /// How long an ambiguous escape-sequence prefix waits for its remaining
796    /// bytes before being forwarded as a real keypress. Terminals send
797    /// sequences in one burst, so only a human typing a lone ESC waits this out.
798    const CARRY_FLUSH_AFTER: Duration = Duration::from_millis(25);
799
800    /// Tracks DEC private mode 2004 (bracketed paste) in the guest's output so
801    /// injected pastes are framed exactly as the terminal would frame a real
802    /// one. Snapshot repaints re-assert tracked modes, so the flag survives
803    /// reattach and heals after any dropped chunk.
804    #[derive(Default)]
805    struct BracketedPasteTracker {
806        tail: Vec<u8>,
807    }
808
809    impl BracketedPasteTracker {
810        fn scan(&mut self, bytes: &[u8], render: &RenderControl) {
811            const INTRO: &[u8] = b"\x1b[?";
812            let mut buf = std::mem::take(&mut self.tail);
813            buf.extend_from_slice(bytes);
814            let mut i = 0;
815            while i < buf.len() {
816                let Some(at) = find(&buf[i..], INTRO) else {
817                    break;
818                };
819                let start = i + at;
820                let mut j = start + INTRO.len();
821                while j < buf.len() && (buf[j].is_ascii_digit() || buf[j] == b';') {
822                    j += 1;
823                }
824                let Some(&fin) = buf.get(j) else {
825                    // Split across chunks: carry the partial sequence, bounded —
826                    // a parameter run longer than any real mode list is not one.
827                    if buf.len() - start <= 24 {
828                        self.tail = buf[start..].to_vec();
829                    }
830                    return;
831                };
832                if fin == b'h' || fin == b'l' {
833                    let in_params = buf[start + INTRO.len()..j]
834                        .split(|&b| b == b';')
835                        .any(|param| param == b"2004");
836                    if in_params {
837                        render.bracketed_paste.store(fin == b'h', Ordering::Relaxed);
838                    }
839                }
840                i = j;
841            }
842            let keep = partial_suffix_len(&buf, INTRO);
843            if keep > 0 {
844                self.tail = buf[buf.len() - keep..].to_vec();
845            }
846        }
847    }
848
849    /// Applies guest-clipboard updates to the local clipboard, so a copy made
850    /// inside the guest is pasteable locally. Exits when the stream ends. The
851    /// clipboard handle stays alive for the whole session: on X11 the
852    /// selection lives only as long as the handle that set it.
853    fn spawn_clipboard_consumer(proc: Arc<ExecProcess>) -> thread::JoinHandle<()> {
854        thread::spawn(move || {
855            let mut clipboard: Option<arboard::Clipboard> = None;
856            while let Some((mime, data)) = block_on(proc.next_clipboard_update()) {
857                if mime != "text/plain" {
858                    continue;
859                }
860                let Ok(text) = String::from_utf8(data) else {
861                    continue;
862                };
863                if clipboard.is_none() {
864                    clipboard = arboard::Clipboard::new().ok();
865                }
866                if let Some(clipboard) = clipboard.as_mut() {
867                    let _ = clipboard.set_text(text);
868                }
869            }
870        })
871    }
872
873    /// The forwarding twin of [`drive_input`]: stdin is scanned for bracketed
874    /// pastes and the Ctrl+V chord (see [`crate::shell_input`]), which the
875    /// [`PasteBridge`] turns into uploads, clipboard pushes, or verbatim
876    /// forwards; every other byte passes through untouched.
877    fn drive_input_forwarding(mut bridge: PasteBridge, stop: &AtomicBool) {
878        let mut scanner = InputScanner::new();
879        let mut buf = [0u8; 4096];
880        let mut stdin_open = true;
881        let mut carry_deadline: Option<Instant> = None;
882        while !stop.load(Ordering::Relaxed) {
883            if RESIZE_PENDING.swap(false, Ordering::Relaxed) {
884                let (cols, rows) = terminal_size();
885                block_on(bridge.proc.resize(cols, rows));
886            }
887            if !stdin_open {
888                thread::sleep(Duration::from_millis(20));
889                continue;
890            }
891            // Keystrokes read while an upload owned stdin replay first, in
892            // order. Refresh the idle deadline like the read path: a replayed
893            // bare ESC lands in the scanner carry and must still flush.
894            if !bridge.stashed_input.is_empty() {
895                let stashed = std::mem::take(&mut bridge.stashed_input);
896                if bridge.handle_events(scanner.scan(&stashed)).is_err() {
897                    break;
898                }
899                carry_deadline = scanner
900                    .has_idle_carry()
901                    .then(|| Instant::now() + CARRY_FLUSH_AFTER);
902            }
903            let n = unsafe {
904                libc::read(
905                    libc::STDIN_FILENO,
906                    buf.as_mut_ptr().cast::<libc::c_void>(),
907                    buf.len(),
908                )
909            };
910            match n.cmp(&0) {
911                std::cmp::Ordering::Greater => {
912                    if bridge
913                        .handle_events(scanner.scan(&buf[..n as usize]))
914                        .is_err()
915                    {
916                        break;
917                    }
918                    carry_deadline = scanner
919                        .has_idle_carry()
920                        .then(|| Instant::now() + CARRY_FLUSH_AFTER);
921                }
922                std::cmp::Ordering::Equal => {
923                    // Local stdin reached EOF: flush whatever the scanner
924                    // still held verbatim, then send EOF and keep draining
925                    // output until the remote process exits.
926                    let held = scanner.flush_all();
927                    if !held.is_empty() && bridge.forward(&held).is_err() {
928                        break;
929                    }
930                    let _ = block_on(bridge.proc.close_stdin());
931                    stdin_open = false;
932                }
933                std::cmp::Ordering::Less => {
934                    let err = std::io::Error::last_os_error();
935                    if !matches!(
936                        err.kind(),
937                        std::io::ErrorKind::WouldBlock | std::io::ErrorKind::Interrupted
938                    ) {
939                        break;
940                    }
941                    // Idle: a carried escape prefix past its deadline was a
942                    // real keypress (e.g. a lone ESC), so forward it now. A
943                    // mid-paste hold is exempt (has_idle_carry): the paste's
944                    // remaining bytes are still coming.
945                    if carry_deadline.is_some_and(|deadline| Instant::now() >= deadline) {
946                        carry_deadline = None;
947                        if scanner.has_idle_carry() {
948                            let carry = scanner.take_carry();
949                            if !carry.is_empty() && bridge.forward(&carry).is_err() {
950                                break;
951                            }
952                        }
953                    }
954                    thread::sleep(Duration::from_millis(10));
955                }
956            }
957        }
958    }
959
960    /// What the local clipboard holds, in the order pasting cares about:
961    /// copied files, then an image (encoded to PNG), then text.
962    enum LocalClipboard {
963        Files(Vec<PathBuf>),
964        Image(Vec<u8>),
965        Text(String),
966        Empty,
967    }
968
969    fn read_local_clipboard() -> LocalClipboard {
970        let Ok(mut clipboard) = arboard::Clipboard::new() else {
971            return LocalClipboard::Empty;
972        };
973        if let Ok(files) = clipboard.get().file_list() {
974            if !files.is_empty() && files.iter().all(|path| path.is_file()) {
975                return LocalClipboard::Files(files);
976            }
977        }
978        if let Ok(image) = clipboard.get_image() {
979            if let Some(png) = encode_png(&image) {
980                return LocalClipboard::Image(png);
981            }
982        }
983        match clipboard.get_text() {
984            Ok(text) if !text.is_empty() => LocalClipboard::Text(text),
985            _ => LocalClipboard::Empty,
986        }
987    }
988
989    /// Encode arboard's raw RGBA image as PNG, the type both the guest
990    /// clipboard and the coding agents (claude, codex) inside it expect. The
991    /// encoder itself rejects a byte buffer that does not match the
992    /// dimensions.
993    fn encode_png(image: &arboard::ImageData) -> Option<Vec<u8>> {
994        let width = u32::try_from(image.width).ok()?;
995        let height = u32::try_from(image.height).ok()?;
996        let mut out = Vec::new();
997        let mut encoder = png::Encoder::new(&mut out, width, height);
998        encoder.set_color(png::ColorType::Rgba);
999        encoder.set_depth(png::BitDepth::Eight);
1000        let mut writer = encoder.write_header().ok()?;
1001        writer.write_image_data(&image.bytes).ok()?;
1002        writer.finish().ok()?;
1003        Some(out)
1004    }
1005
1006    /// How an upload ended: the guest paths to paste, a user cancel (paste
1007    /// nothing), or a failure (the caller falls back to forwarding the
1008    /// original bytes where that makes sense).
1009    enum UploadOutcome {
1010        Done(Vec<String>),
1011        Cancelled,
1012        Failed,
1013    }
1014
1015    /// One file (or in-memory blob) headed for the guest drops directory.
1016    struct UploadSource {
1017        guest_path: String,
1018        size: u64,
1019        data: UploadData,
1020    }
1021
1022    enum UploadData {
1023        File(PathBuf),
1024        Memory(Vec<u8>),
1025    }
1026
1027    /// Turns scanned paste events into guest activity: dragged files upload
1028    /// and paste as guest paths, Ctrl+V forwards the local clipboard (guest
1029    /// clipboard when supported, file upload otherwise), everything else
1030    /// forwards verbatim.
1031    struct PasteBridge {
1032        proc: Arc<ExecProcess>,
1033        render: Arc<RenderControl>,
1034        /// This session's private drop directory under GUEST_DROPS_ROOT, keyed
1035        /// by the exec id so concurrent shells never share a path.
1036        drops_dir: String,
1037        /// Guest file names already used this session, so a re-dropped name
1038        /// gets a numbered variant instead of clobbering a different file.
1039        used_names: std::collections::HashSet<String>,
1040        /// Whether this guest accepts clipboard writes. `None` until the first
1041        /// attempt; latches `Some(false)` on Unimplemented (the guest has no
1042        /// clipboard) so later pastes skip straight to the file fallback.
1043        guest_clipboard: Option<bool>,
1044        /// Keystrokes read while an upload owned stdin (watching for cancel),
1045        /// replayed in order afterwards.
1046        stashed_input: Vec<u8>,
1047    }
1048
1049    impl PasteBridge {
1050        fn new(proc: Arc<ExecProcess>, render: Arc<RenderControl>) -> PasteBridge {
1051            // Name the per-session directory by the hash of the exec id, so it
1052            // is always path-safe, never `.`/`..`, and collision-free whatever
1053            // shape the server's id takes: two concurrent sessions never share
1054            // a path.
1055            use sha2::Digest as _;
1056            use std::fmt::Write as _;
1057            let digest = sha2::Sha256::digest(proc.exec_request_id().as_bytes());
1058            let mut session = String::with_capacity(16);
1059            for byte in &digest[..8] {
1060                let _ = write!(session, "{byte:02x}");
1061            }
1062            let drops_dir = format!("{GUEST_DROPS_ROOT}/{session}");
1063            PasteBridge {
1064                proc,
1065                render,
1066                drops_dir,
1067                used_names: std::collections::HashSet::new(),
1068                guest_clipboard: None,
1069                stashed_input: Vec::new(),
1070            }
1071        }
1072
1073        fn handle_events(&mut self, events: Vec<InputEvent>) -> Result<(), ()> {
1074            for event in events {
1075                match event {
1076                    InputEvent::Bytes(bytes) => self.forward(&bytes)?,
1077                    InputEvent::Paste(body) => self.handle_paste(&body)?,
1078                    InputEvent::PasteChord(chord) => self.handle_chord(&chord)?,
1079                }
1080            }
1081            Ok(())
1082        }
1083
1084        /// Forward bytes to the guest pty. Err means the exec ended and the
1085        /// input loop should stop, matching [`drive_input`].
1086        fn forward(&self, bytes: &[u8]) -> Result<(), ()> {
1087            block_on(self.proc.write_stdin(bytes)).map_err(|_| ())
1088        }
1089
1090        /// A completed bracketed paste: a drag-and-drop of local files uploads
1091        /// them and pastes the guest paths; any other paste (or a failed
1092        /// upload) forwards byte-identically.
1093        fn handle_paste(&mut self, body: &[u8]) -> Result<(), ()> {
1094            if let Some(files) = dropped_local_files(body) {
1095                match self.upload_files(&files) {
1096                    UploadOutcome::Done(paths) => {
1097                        // Keep the trailing separator the drag arrived with,
1098                        // so typing right after the drop stays a separate
1099                        // argument exactly as it would locally.
1100                        let text = format!("{} ", paths.join(" "));
1101                        return self.inject_uploaded(&text, &paths);
1102                    }
1103                    UploadOutcome::Cancelled => return Ok(()),
1104                    UploadOutcome::Failed => {} // fall through to the original paste
1105                }
1106            }
1107            let mut raw = Vec::with_capacity(body.len() + PASTE_START.len() + PASTE_END.len());
1108            raw.extend_from_slice(PASTE_START);
1109            raw.extend_from_slice(body);
1110            raw.extend_from_slice(PASTE_END);
1111            self.forward(&raw)
1112        }
1113
1114        /// A Ctrl+V press: make the local clipboard available in the guest,
1115        /// then (except for uploads that paste a path themselves) deliver the
1116        /// keypress so the guest application reacts to it as usual.
1117        fn handle_chord(&mut self, chord: &[u8]) -> Result<(), ()> {
1118            match read_local_clipboard() {
1119                LocalClipboard::Files(files) => {
1120                    let outcome = self.upload_files(&files);
1121                    self.inject_outcome(outcome)
1122                }
1123                LocalClipboard::Image(png) => {
1124                    if png.len() <= CLIPBOARD_PUSH_MAX && self.push_clipboard("image/png", &png) {
1125                        return self.forward(chord);
1126                    }
1127                    // No guest clipboard took the image (a non-devbox guest, or
1128                    // one over the push cap): upload it and inject the path, the
1129                    // only way a paste-reading agent gets an image here. The
1130                    // chord is deliberately not forwarded — doing so would paste
1131                    // whatever the guest clipboard last held. Unlike text, which
1132                    // is on the clipboard almost always (so the text path never
1133                    // injects, to keep vim visual-block / quoted-insert intact),
1134                    // an image on the clipboard is almost always an intended
1135                    // paste, so injecting wins over preserving a Ctrl+V binding.
1136                    let name = self.reserve_name(std::path::Path::new("clipboard.png"));
1137                    let outcome = self.upload_bytes(name, png);
1138                    self.inject_outcome(outcome)
1139                }
1140                LocalClipboard::Text(text) => {
1141                    if text.len() <= CLIPBOARD_PUSH_MAX {
1142                        // The chord is forwarded whether or not the push
1143                        // lands. Ctrl+V is not only "paste": vim binds it to
1144                        // visual-block and readline to quoted-insert, and text
1145                        // sits on the clipboard almost always, so replacing a
1146                        // failed push with injected text or a file would fire
1147                        // inside those apps constantly. On a guest with no
1148                        // clipboard the app's paste read finds nothing, which
1149                        // is exactly how these sessions behaved before
1150                        // clipboard forwarding existed; terminal-level paste
1151                        // (Cmd+V) remains the text path there.
1152                        self.push_clipboard("text/plain", text.as_bytes());
1153                        return self.forward(chord);
1154                    }
1155                    // Too big for a clipboard push. Forwarding the keypress
1156                    // anyway would paste whatever the guest clipboard last
1157                    // held, so deliver the text as a file like an oversized
1158                    // image.
1159                    let name = self.reserve_name(std::path::Path::new("clipboard.txt"));
1160                    let outcome = self.upload_bytes(name, text.into_bytes());
1161                    self.inject_outcome(outcome)
1162                }
1163                LocalClipboard::Empty => self.forward(chord),
1164            }
1165        }
1166
1167        /// Try to place content on the guest clipboard, reporting success,
1168        /// waiting at most [`CLIPBOARD_PUSH_DEADLINE`]. Unimplemented latches
1169        /// the fallback: this guest has no clipboard (its image ships none,
1170        /// or it predates the feature), and that never changes mid-session.
1171        /// Other failures, the deadline included, just skip the push this
1172        /// time.
1173        fn push_clipboard(&mut self, mime: &str, data: &[u8]) -> bool {
1174            if self.guest_clipboard == Some(false) {
1175                return false;
1176            }
1177            let pushed = block_on(async {
1178                tokio::time::timeout(CLIPBOARD_PUSH_DEADLINE, self.proc.set_clipboard(mime, data))
1179                    .await
1180            });
1181            match pushed {
1182                Ok(Ok(())) => {
1183                    self.guest_clipboard = Some(true);
1184                    true
1185                }
1186                Ok(Err(SailError::Execution {
1187                    code: RpcStatus::Unimplemented,
1188                    ..
1189                })) => {
1190                    self.guest_clipboard = Some(false);
1191                    false
1192                }
1193                Ok(Err(_)) | Err(_) => false,
1194            }
1195        }
1196
1197        /// Paste an upload's guest paths, or forward nothing when the upload was
1198        /// cancelled or failed (the original chord/paste was already handled).
1199        fn inject_outcome(&mut self, outcome: UploadOutcome) -> Result<(), ()> {
1200            match outcome {
1201                UploadOutcome::Done(paths) => self.inject_uploaded(&paths.join(" "), &paths),
1202                UploadOutcome::Cancelled | UploadOutcome::Failed => Ok(()),
1203            }
1204        }
1205
1206        /// Paste the uploaded files' guest paths, deleting the uploads if the
1207        /// session ends before the paste can land, so nothing stays
1208        /// unreferenced in the guest.
1209        fn inject_uploaded(&mut self, text: &str, paths: &[String]) -> Result<(), ()> {
1210            if self.inject(text).is_err() {
1211                let _ = block_on(self.proc.remove_guest_files(paths));
1212                return Err(());
1213            }
1214            Ok(())
1215        }
1216
1217        /// Paste text into the guest exactly as the terminal would: bracketed
1218        /// while the application has mode 2004 on, plain keystrokes otherwise.
1219        fn inject(&self, text: &str) -> Result<(), ()> {
1220            let bracketed = self.render.bracketed_paste.load(Ordering::Relaxed);
1221            let mut bytes = Vec::with_capacity(text.len() + 16);
1222            if bracketed {
1223                bytes.extend_from_slice(PASTE_START);
1224            }
1225            bytes.extend_from_slice(text.as_bytes());
1226            if bracketed {
1227                bytes.extend_from_slice(PASTE_END);
1228            }
1229            self.forward(&bytes)
1230        }
1231
1232        /// Reserve a guest-side name for an upload, numbering repeats
1233        /// (photo.png, photo-2.png, ...) within the session.
1234        fn reserve_name(&mut self, path: &std::path::Path) -> String {
1235            let base = sanitize_drop_name(path);
1236            let mut name = base.clone();
1237            let mut n = 1;
1238            while !self.used_names.insert(name.clone()) {
1239                n += 1;
1240                name = match base.rsplit_once('.') {
1241                    Some((stem, ext)) if !stem.is_empty() => format!("{stem}-{n}.{ext}"),
1242                    _ => format!("{base}-{n}"),
1243                };
1244            }
1245            name
1246        }
1247
1248        fn upload_files(&mut self, files: &[PathBuf]) -> UploadOutcome {
1249            let mut sources = Vec::with_capacity(files.len());
1250            for path in files {
1251                let Ok(meta) = std::fs::metadata(path) else {
1252                    return UploadOutcome::Failed;
1253                };
1254                let name = self.reserve_name(path);
1255                sources.push(UploadSource {
1256                    guest_path: format!("{}/{name}", self.drops_dir),
1257                    size: meta.len(),
1258                    data: UploadData::File(path.clone()),
1259                });
1260            }
1261            self.run_upload(sources)
1262        }
1263
1264        /// Upload one in-memory blob under a name already reserved via
1265        /// [`reserve_name`](Self::reserve_name).
1266        fn upload_bytes(&mut self, name: String, bytes: Vec<u8>) -> UploadOutcome {
1267            self.run_upload(vec![UploadSource {
1268                guest_path: format!("{}/{name}", self.drops_dir),
1269                size: bytes.len() as u64,
1270                data: UploadData::Memory(bytes),
1271            }])
1272        }
1273
1274        /// Stream the sources to the guest while this thread keeps the
1275        /// terminal responsive: a progress line appears for slow uploads, Esc
1276        /// or Ctrl+C cancels (aborting the write streams, which the guest
1277        /// discards uncommitted), and other keystrokes are stashed for replay.
1278        /// Guest output rendering is paused throughout; if anything was drawn
1279        /// over the screen, a resync repaints it from the authoritative guest
1280        /// screen state.
1281        fn run_upload(&mut self, sources: Vec<UploadSource>) -> UploadOutcome {
1282            let total: u64 = sources.iter().map(|s| s.size).sum();
1283            let guest_paths: Vec<String> = sources.iter().map(|s| s.guest_path.clone()).collect();
1284            let label = if sources.len() == 1 {
1285                guest_paths[0]
1286                    .rsplit('/')
1287                    .next()
1288                    .unwrap_or_default()
1289                    .to_string()
1290            } else {
1291                format!("{} files", sources.len())
1292            };
1293            self.render.paused.store(true, Ordering::Relaxed);
1294            let progress = Arc::new(std::sync::atomic::AtomicU64::new(0));
1295            let committed: Arc<std::sync::Mutex<Vec<String>>> =
1296                Arc::new(std::sync::Mutex::new(Vec::new()));
1297            // Spawn where the exec stream lives (see block_on): an embedding
1298            // runtime's channels must not be redialed on the crate's own
1299            // reactor.
1300            let handle = tokio::runtime::Handle::try_current()
1301                .unwrap_or_else(|_| crate::runtime::runtime().handle().clone());
1302            let mut task = handle.spawn(upload_task(
1303                Arc::clone(&self.proc),
1304                sources,
1305                Arc::clone(&progress),
1306                Arc::clone(&committed),
1307            ));
1308            let started = Instant::now();
1309            let mut ui = ProgressLine::new(label, total);
1310            let mut esc_at: Option<Instant> = None;
1311            let outcome = loop {
1312                if task.is_finished() {
1313                    // A bare ESC still inside its disambiguation window was a
1314                    // real keypress after all; replay it with the other
1315                    // stashed input instead of dropping it.
1316                    if esc_at.take().is_some() {
1317                        self.stashed_input.push(0x1b);
1318                    }
1319                    break match block_on(&mut task) {
1320                        Ok(Ok(())) => UploadOutcome::Done(guest_paths.clone()),
1321                        Ok(Err(err)) => {
1322                            self.roll_back_upload(&committed, &guest_paths);
1323                            ui.flash(&format!("[sail] upload failed: {err}"));
1324                            UploadOutcome::Failed
1325                        }
1326                        Err(_) => {
1327                            self.roll_back_upload(&committed, &guest_paths);
1328                            UploadOutcome::Failed
1329                        }
1330                    };
1331                }
1332                if self.poll_cancel(&mut esc_at) {
1333                    // Aborting drops the in-flight writer mid-stream; the
1334                    // guest treats the torn stream as uncommitted and discards
1335                    // it.
1336                    task.abort();
1337                    let _ = block_on(&mut task);
1338                    self.roll_back_upload(&committed, &guest_paths);
1339                    ui.flash("[sail] upload canceled");
1340                    break UploadOutcome::Cancelled;
1341                }
1342                ui.tick(started, progress.load(Ordering::Relaxed));
1343                thread::sleep(Duration::from_millis(30));
1344            };
1345            ui.clear();
1346            self.render.paused.store(false, Ordering::Relaxed);
1347            if ui.wrote {
1348                // The progress line scribbled over the guest's screen; repaint
1349                // it from the guest's authoritative screen state.
1350                block_on(self.proc.resync());
1351            }
1352            outcome
1353        }
1354
1355        /// Undo an upload that will paste nothing (cancelled or failed):
1356        /// delete the files that already committed — nothing points at them,
1357        /// and cancel means the user wants none of the dragged content in the
1358        /// guest — and release the reserved names so a retried drag lands on
1359        /// the same paths. Deletion is best effort: if the guest is unreachable
1360        /// the unreferenced files linger in its /tmp until the Sailbox goes away.
1361        fn roll_back_upload(
1362            &mut self,
1363            committed: &Arc<std::sync::Mutex<Vec<String>>>,
1364            guest_paths: &[String],
1365        ) {
1366            let committed = std::mem::take(&mut *committed.lock().unwrap());
1367            if !committed.is_empty() {
1368                let _ = block_on(self.proc.remove_guest_files(&committed));
1369            }
1370            for path in guest_paths {
1371                if let Some(name) = path.rsplit('/').next() {
1372                    self.used_names.remove(name);
1373                }
1374            }
1375        }
1376
1377        /// Drain stdin during an upload. Ctrl+C cancels at once; a bare ESC
1378        /// cancels after a short pause — long enough for the rest of an escape
1379        /// sequence (an arrow key) to arrive and be stashed instead. Everything
1380        /// else is stashed and replayed after the upload.
1381        fn poll_cancel(&mut self, esc_at: &mut Option<Instant>) -> bool {
1382            let mut buf = [0u8; 256];
1383            loop {
1384                let n = unsafe {
1385                    libc::read(
1386                        libc::STDIN_FILENO,
1387                        buf.as_mut_ptr().cast::<libc::c_void>(),
1388                        buf.len(),
1389                    )
1390                };
1391                if n <= 0 {
1392                    break;
1393                }
1394                let bytes = &buf[..n as usize];
1395                for (idx, &byte) in bytes.iter().enumerate() {
1396                    if byte == 0x03 {
1397                        // Keys typed in the same burst as the cancel replay
1398                        // after the upload settles.
1399                        self.stashed_input.extend_from_slice(&bytes[idx + 1..]);
1400                        return true;
1401                    }
1402                    if esc_at.take().is_some() {
1403                        // The pending ESC was the start of a sequence after all.
1404                        self.stashed_input.push(0x1b);
1405                    }
1406                    if byte == 0x1b && idx == bytes.len() - 1 {
1407                        *esc_at = Some(Instant::now());
1408                    } else {
1409                        self.stashed_input.push(byte);
1410                    }
1411                }
1412            }
1413            // 60ms: longer than one 30ms poll tick of run_upload, so a split
1414            // escape sequence has a whole further poll to finish arriving.
1415            esc_at.is_some_and(|at| at.elapsed() >= Duration::from_millis(60))
1416        }
1417    }
1418
1419    /// The background half of an upload: stream every source into the guest,
1420    /// publishing progress for the interactive thread's UI and each committed
1421    /// guest path for cancel's rollback.
1422    async fn upload_task(
1423        proc: Arc<ExecProcess>,
1424        sources: Vec<UploadSource>,
1425        progress: Arc<std::sync::atomic::AtomicU64>,
1426        committed: Arc<std::sync::Mutex<Vec<String>>>,
1427    ) -> Result<(), SailError> {
1428        use tokio::io::AsyncReadExt;
1429        for source in sources {
1430            let mut writer = proc.guest_file_writer(&source.guest_path);
1431            match source.data {
1432                UploadData::Memory(bytes) => {
1433                    for chunk in bytes.chunks(UPLOAD_CHUNK_BYTES) {
1434                        writer.write_chunk(chunk.to_vec()).await?;
1435                        progress.fetch_add(chunk.len() as u64, Ordering::Relaxed);
1436                    }
1437                }
1438                UploadData::File(path) => {
1439                    let mut file =
1440                        tokio::fs::File::open(&path)
1441                            .await
1442                            .map_err(|err| SailError::Internal {
1443                                message: format!("read {}: {err}", path.display()),
1444                            })?;
1445                    let mut chunk = vec![0u8; UPLOAD_CHUNK_BYTES];
1446                    loop {
1447                        let n = file
1448                            .read(&mut chunk)
1449                            .await
1450                            .map_err(|err| SailError::Internal {
1451                                message: format!("read {}: {err}", path.display()),
1452                            })?;
1453                        if n == 0 {
1454                            break;
1455                        }
1456                        writer.write_chunk(chunk[..n].to_vec()).await?;
1457                        progress.fetch_add(n as u64, Ordering::Relaxed);
1458                    }
1459                }
1460            }
1461            // Record the path before finish()'s await: finish closes the
1462            // client stream, so a cancel that aborts this task during the
1463            // await can still let the guest commit the file. Recording first
1464            // guarantees rollback has the path — an rm of a file that never
1465            // committed is a harmless no-op.
1466            committed.lock().unwrap().push(source.guest_path);
1467            writer.finish().await?;
1468        }
1469        Ok(())
1470    }
1471
1472    /// The one-line upload status drawn at the cursor. It only ever repaints
1473    /// itself in place; whatever it overwrote is restored by the post-upload
1474    /// resync.
1475    struct ProgressLine {
1476        label: String,
1477        total: u64,
1478        wrote: bool,
1479        last_draw: Option<Instant>,
1480    }
1481
1482    impl ProgressLine {
1483        fn new(label: String, total: u64) -> ProgressLine {
1484            ProgressLine {
1485                label,
1486                total,
1487                wrote: false,
1488                last_draw: None,
1489            }
1490        }
1491
1492        fn tick(&mut self, started: Instant, sent: u64) {
1493            if !self.wrote && started.elapsed() < UPLOAD_UI_DELAY {
1494                return;
1495            }
1496            if self
1497                .last_draw
1498                .is_some_and(|last| last.elapsed() < Duration::from_millis(100))
1499            {
1500                return;
1501            }
1502            self.last_draw = Some(Instant::now());
1503            self.wrote = true;
1504            let percent = (sent.min(self.total) * 100)
1505                .checked_div(self.total)
1506                .unwrap_or(100);
1507            write_terminal_line(&format!(
1508                "[sail] uploading {}  {percent}%  {} / {}  (esc cancels)",
1509                self.label,
1510                format_bytes(sent),
1511                format_bytes(self.total),
1512            ));
1513        }
1514
1515        /// Show a final status long enough to read before the screen repaints.
1516        fn flash(&mut self, message: &str) {
1517            self.wrote = true;
1518            write_terminal_line(message);
1519            thread::sleep(Duration::from_millis(1200));
1520        }
1521
1522        fn clear(&mut self) {
1523            if self.wrote {
1524                write_terminal(ERASE_LINE);
1525            }
1526        }
1527    }
1528
1529    /// Carriage return + erase-line: return to column 0 and clear the row, so
1530    /// the next write overwrites the cursor line in place.
1531    const ERASE_LINE: &[u8] = b"\r\x1b[2K";
1532
1533    /// Overwrite the cursor line with `text`.
1534    fn write_terminal_line(text: &str) {
1535        let mut bytes = Vec::with_capacity(text.len() + ERASE_LINE.len());
1536        bytes.extend_from_slice(ERASE_LINE);
1537        bytes.extend_from_slice(text.as_bytes());
1538        write_terminal(&bytes);
1539    }
1540
1541    fn write_terminal(bytes: &[u8]) {
1542        let mut sink = RawFdWriter(libc::STDOUT_FILENO);
1543        flush_blocking(&mut sink, bytes);
1544    }
1545
1546    /// Sizes for the progress line, in the 1000-based MB/KB a file manager
1547    /// labels the dragged file with.
1548    fn format_bytes(n: u64) -> String {
1549        const MB: u64 = 1_000_000;
1550        if n >= 10 * MB {
1551            format!("{} MB", n / MB)
1552        } else if n >= MB {
1553            format!("{:.1} MB", n as f64 / MB as f64)
1554        } else {
1555            format!("{} KB", n.div_ceil(1000))
1556        }
1557    }
1558
1559    // --- platform terminal plumbing ---
1560
1561    /// Put the local terminal into raw mode, returning the saved settings.
1562    fn enter_raw_mode() -> Result<libc::termios, SailError> {
1563        unsafe {
1564            let mut saved: libc::termios = std::mem::zeroed();
1565            if libc::tcgetattr(libc::STDIN_FILENO, &raw mut saved) != 0 {
1566                return Err(SailError::Internal {
1567                    message: format!(
1568                        "could not enter raw terminal mode: {}",
1569                        std::io::Error::last_os_error()
1570                    ),
1571                });
1572            }
1573            let mut raw = saved;
1574            libc::cfmakeraw(&raw mut raw);
1575            if libc::tcsetattr(libc::STDIN_FILENO, libc::TCSADRAIN, &raw const raw) != 0 {
1576                return Err(SailError::Internal {
1577                    message: format!(
1578                        "could not enter raw terminal mode: {}",
1579                        std::io::Error::last_os_error()
1580                    ),
1581                });
1582            }
1583            Ok(saved)
1584        }
1585    }
1586
1587    fn restore_terminal(saved: &libc::termios) {
1588        unsafe {
1589            let _ = libc::tcsetattr(
1590                libc::STDIN_FILENO,
1591                libc::TCSADRAIN,
1592                std::ptr::from_ref(saved),
1593            );
1594        }
1595    }
1596
1597    type SigHandler = libc::sighandler_t;
1598
1599    fn install_sigwinch() -> SigHandler {
1600        // `signal` takes the handler as a numeric `sighandler_t`; cast through a
1601        // concrete fn pointer first so this is a pointer-to-int cast, not a
1602        // fn-item-to-int cast.
1603        let handler = on_sigwinch as extern "C" fn(libc::c_int) as usize;
1604        unsafe { libc::signal(libc::SIGWINCH, handler) }
1605    }
1606
1607    fn restore_sigwinch(prev: SigHandler) {
1608        unsafe {
1609            libc::signal(libc::SIGWINCH, prev);
1610        }
1611    }
1612
1613    /// Put stdin into non-blocking mode so the input loop can interleave reads
1614    /// with resize handling and the stop flag. Returns the previous fcntl flags.
1615    fn set_stdin_nonblocking() -> libc::c_int {
1616        unsafe {
1617            let flags = libc::fcntl(libc::STDIN_FILENO, libc::F_GETFL);
1618            if flags >= 0 {
1619                libc::fcntl(libc::STDIN_FILENO, libc::F_SETFL, flags | libc::O_NONBLOCK);
1620            }
1621            flags
1622        }
1623    }
1624
1625    fn restore_stdin_flags(flags: libc::c_int) {
1626        if flags >= 0 {
1627            unsafe {
1628                libc::fcntl(libc::STDIN_FILENO, libc::F_SETFL, flags);
1629            }
1630        }
1631    }
1632
1633    /// Put stdout into non-blocking mode so the output pump is never parked in a
1634    /// write to a slow terminal. Returns the previous fcntl flags.
1635    fn set_stdout_nonblocking() -> libc::c_int {
1636        unsafe {
1637            let flags = libc::fcntl(libc::STDOUT_FILENO, libc::F_GETFL);
1638            if flags >= 0 {
1639                libc::fcntl(libc::STDOUT_FILENO, libc::F_SETFL, flags | libc::O_NONBLOCK);
1640            }
1641            flags
1642        }
1643    }
1644
1645    fn restore_stdout_flags(flags: libc::c_int) {
1646        if flags >= 0 {
1647            unsafe {
1648                libc::fcntl(libc::STDOUT_FILENO, libc::F_SETFL, flags);
1649            }
1650        }
1651    }
1652
1653    #[cfg(test)]
1654    mod tests {
1655        use super::*;
1656
1657        #[test]
1658        fn resync_due_throttles_back_to_back_requests() {
1659            let mut last = None;
1660            // The first request is always due and stamps the clock.
1661            assert!(resync_due(&mut last));
1662            // A second request within RESYNC_MIN_INTERVAL is suppressed, so a
1663            // persistently-behind reader cannot flood the guest with resync RPCs.
1664            assert!(!resync_due(&mut last));
1665        }
1666
1667        #[test]
1668        fn format_bytes_matches_its_thousand_based_labels() {
1669            assert_eq!(format_bytes(0), "0 KB");
1670            assert_eq!(format_bytes(1), "1 KB");
1671            assert_eq!(format_bytes(999_999), "1000 KB");
1672            assert_eq!(format_bytes(1_000_000), "1.0 MB");
1673            assert_eq!(format_bytes(3_200_000), "3.2 MB");
1674            assert_eq!(format_bytes(25_000_000), "25 MB");
1675        }
1676    }
1677}
1678
1679#[cfg(test)]
1680mod tests {
1681    use super::*;
1682
1683    #[test]
1684    fn login_shell_quotes_an_explicit_path() {
1685        // A path with spaces runs as one literal program.
1686        assert_eq!(
1687            login_shell_command(Some("/opt/my tools/zsh")),
1688            "exec '/opt/my tools/zsh' -l"
1689        );
1690        // The default stays unquoted so the guest expands $SHELL.
1691        assert_eq!(
1692            login_shell_command(/* shell */ None),
1693            "exec ${SHELL:-/bin/bash} -l"
1694        );
1695    }
1696
1697    #[tokio::test]
1698    async fn shell_requires_a_tty() {
1699        // Test processes have no TTY on stdin/stdout, so the precondition
1700        // fires before any network or terminal manipulation.
1701        let client = crate::Client::builder("sk_test")
1702            .api_url("http://127.0.0.1:1")
1703            .sailbox_api_url("http://127.0.0.1:1")
1704            .build()
1705            .expect("build");
1706        let err = client
1707            .sailbox("sb_test")
1708            .shell(/* command */ None, ShellOptions::default())
1709            .await
1710            .expect_err("no tty in tests");
1711        assert!(err.to_string().contains("interactive terminal"), "{err}");
1712    }
1713}