sail-rs 0.2.11

//! Streaming exec engine: a running command in a sailbox with live output.
//!
//! [`ExecProcess::start`] opens the worker proxy's server-streaming
//! `StreamSailboxExec` RPC. The first frame is `Started` (carrying the durable
//! `exec_request_id`); chunk frames carry live stdout/stderr; a terminal Exit
//! frame carries the result. The command is detached on the server, so dropping
//! the handle never kills it.
//!
//! A background pump task drains the stream into two drop-oldest ring buffers
//! that synchronous readers ([`StreamReader`]) pull from. If the stream breaks
//! mid-run, the pump reconnects with the same idempotency key and the per-stream
//! seqs it last saw, so the guest replays only the unseen tail. Only when the
//! stream cannot be resumed does [`ExecProcess::wait`] fall back to polling
//! `WaitSailboxExec` for the authoritative buffered result. Stdin rides a
//! separate offset-idempotent unary RPC, independent of the output stream.

use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Condvar, Mutex};
use std::time::Duration;

use serde::Serialize;
use tokio::sync::Mutex as AsyncMutex;
use tokio::sync::Notify;
use tonic::{Code, Status, Streaming};

use crate::error::SailError;
use crate::pb::workerproxy::v1 as pb;
use crate::worker::{
    retry_deadline, rpc_attempt_timeout, should_invalidate_channel,
    should_retry_transient_exec_rpc, sleep_before_retry, WorkerProxy,
    EXEC_TRANSIENT_RETRY_INITIAL_DELAY_SECONDS,
};

/// Default budget for transient-RPC retries against a waking/migrating sailbox;
/// the default value of [`ExecOptions::retry_timeout`].
pub const EXEC_TRANSIENT_RETRY_TIMEOUT_SECONDS: f64 = 30.0;
/// Local cap on buffered stream output: a slow reader loses the oldest output
/// rather than blocking the stream. Sized in UTF-8 bytes to the server's
/// in-memory exec replay ring so a locally resolved tail is the same size the
/// server replays on a reattach. A backend test keeps this in lockstep with the
/// ring (`guestExecChunkBufferBytes`); change both together.
const STREAM_BUFFER_CAP_BYTES: usize = 1024 * 1024;
/// Stdin writes are chunked so a single RPC stays well under gRPC message limits
/// and partial accepts resume cheaply.
const STDIN_WRITE_CHUNK_BYTES: usize = 256 * 1024;

/// Lock a mutex, recovering the guard if a peer panicked while holding it
/// (matching `channels.rs`). The data under these locks is simple, so a poisoned
/// peer should degrade rather than cascade a panic into reader/pump threads.
fn lock<T>(mutex: &Mutex<T>) -> std::sync::MutexGuard<'_, T> {
    mutex
        .lock()
        .unwrap_or_else(std::sync::PoisonError::into_inner)
}

/// Which output stream a chunk or reader belongs to.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum OutputStream {
    /// The standard output stream.
    Stdout,
    /// The standard error stream.
    Stderr,
}

/// One step of reading a live output stream.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ReadStep {
    /// The next retained chunk of output.
    Chunk(String),
    /// The stream is closed and fully drained.
    Eof,
    /// Nothing new before the timeout; the caller may check for signals and
    /// retry.
    Pending,
}

/// The buffered result of a finished exec.
#[derive(Debug, Clone, Serialize)]
#[non_exhaustive]
#[allow(clippy::struct_excessive_bools)]
pub struct ExecResult {
    /// Buffered stdout, decoded as UTF-8.
    pub stdout: String,
    /// Buffered stderr, decoded as UTF-8.
    pub stderr: String,
    /// The command's exit code.
    pub exit_code: i32,
    /// Whether the command was killed for exceeding its timeout.
    pub timed_out: bool,
    /// Whether stdout overflowed the server output ring and lost its oldest
    /// bytes.
    pub stdout_truncated: bool,
    /// Whether stderr overflowed the server output ring and lost its oldest
    /// bytes.
    pub stderr_truncated: bool,
    /// Whether the live stream delivered stdout through to the command's exit.
    /// When true, a consumer that streamed the output live already holds the
    /// complete stdout even if `stdout` here is a truncated buffered tail. When
    /// false (the stream ended before the exit, or no exit was observed),
    /// `stdout` is the authoritative buffered copy to fall back on.
    pub stdout_complete: bool,
    /// Whether the live stream delivered stderr through to the command's exit
    /// (see `stdout_complete`).
    pub stderr_complete: bool,
}

/// Which signal to send when cancelling a running exec.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CancelSignal {
    /// SIGINT: ask the command to stop (what a first Ctrl-C sends).
    Interrupt,
    /// SIGKILL: force-kill a command that ignored the interrupt.
    Kill,
}

impl CancelSignal {
    /// Whether this is the forceful (SIGKILL) variant, as the wire encodes it.
    fn is_force(self) -> bool {
        matches!(self, CancelSignal::Kill)
    }
}

/// How long to keep retrying transient RPCs against a waking or migrating
/// sailbox before giving up.
#[derive(Debug, Clone, Copy, PartialEq)]
pub enum RetryBudget {
    /// Do not retry; fail on the first transient error.
    None,
    /// Retry for at most this long.
    Within(Duration),
    /// Retry indefinitely, until the call succeeds or hits a non-transient error.
    Forever,
}

impl RetryBudget {
    /// Encode as the seconds the core's retry loop expects: `0` = none, a finite
    /// count = a bounded budget, `+inf` = forever.
    pub fn as_secs_f64(self) -> f64 {
        match self {
            RetryBudget::None => 0.0,
            RetryBudget::Within(d) => d.as_secs_f64(),
            RetryBudget::Forever => f64::INFINITY,
        }
    }

    /// Decode from seconds at the FFI boundary (Python passes an `f64`): `<= 0` =
    /// none, a non-finite value = forever, otherwise a bounded budget.
    pub fn from_secs_f64(secs: f64) -> RetryBudget {
        if secs <= 0.0 {
            RetryBudget::None
        } else if secs.is_finite() {
            RetryBudget::Within(Duration::from_secs_f64(secs))
        } else {
            RetryBudget::Forever
        }
    }
}

/// Optional settings for [`Sailbox::exec`](crate::Sailbox::exec) and
/// [`Sailbox::exec_shell`](crate::Sailbox::exec_shell). `Default` runs a
/// plain foreground command (no pty, no stdin, no timeout) and retries transient
/// failures against a waking or migrating sailbox for 30 seconds (see
/// [`retry_timeout`](Self::retry_timeout)).
#[derive(Debug, Clone)]
pub struct ExecOptions {
    /// Wall-clock limit before the server kills the command; `None` means no
    /// limit. The wire is whole seconds, so a set sub-second timeout rounds up
    /// to 1 second (it never collapses to the no-limit `0`).
    pub timeout: Option<Duration>,
    /// Leave the command's stdin open for [`ExecProcess::write_stdin`].
    pub open_stdin: bool,
    /// Allocate a pseudo-terminal for the command.
    pub pty: bool,
    /// TERM value for the pty (e.g. `xterm-256color`); ignored without `pty`.
    pub term: String,
    /// Initial pty width in columns; ignored without `pty`.
    pub cols: u32,
    /// Initial pty height in rows; ignored without `pty`.
    pub rows: u32,
    /// Stable key that dedupes the launch so a reconnect reattaches to the same
    /// command. Empty mints a fresh one per call.
    pub idempotency_key: String,
    /// Budget for retrying transient RPC failures against a waking or migrating
    /// sailbox: both while opening the stream and when [`ExecProcess::wait`]
    /// falls back to `WaitSailboxExec`, which reattaches to the guest for the
    /// result.
    pub retry_timeout: RetryBudget,
    /// Working directory to run a shell command in. Only valid with
    /// [`Sailbox::exec_shell`](crate::Sailbox::exec_shell).
    pub cwd: Option<String>,
    /// Detach a shell command so it keeps running and the call returns
    /// immediately; output is discarded. Only valid with
    /// [`Sailbox::exec_shell`](crate::Sailbox::exec_shell), and incompatible with
    /// `open_stdin` and `pty`.
    pub background: bool,
}

impl Default for ExecOptions {
    fn default() -> ExecOptions {
        ExecOptions {
            timeout: None,
            open_stdin: false,
            pty: false,
            term: String::new(),
            cols: 0,
            rows: 0,
            idempotency_key: String::new(),
            retry_timeout: RetryBudget::Within(Duration::from_secs_f64(
                EXEC_TRANSIENT_RETRY_TIMEOUT_SECONDS,
            )),
            cwd: None,
            background: false,
        }
    }
}

/// POSIX single-quote a string for safe inclusion in a shell command.
pub(crate) fn sh_quote(value: &str) -> String {
    format!("'{}'", value.replace('\'', "'\\''"))
}

/// Build the `argv` that runs `command` via `/bin/sh -lc`, applying the
/// `cwd`/`background` shell conveniences from `options` and validating their
/// combinations. This is the single implementation behind every SDK's
/// string-command exec.
pub fn shell_argv(command: &str, options: &ExecOptions) -> Result<Vec<String>, SailError> {
    let invalid = |message: &str| {
        Err(SailError::InvalidArgument {
            message: message.to_string(),
        })
    };
    if command.is_empty() {
        return invalid("command must be non-empty");
    }
    if options.background && (options.open_stdin || options.pty) {
        return invalid("background is not supported with open_stdin or pty");
    }
    let mut command = command.to_string();
    if let Some(cwd) = &options.cwd {
        let cwd = cwd.trim();
        if cwd.is_empty() {
            return invalid("cwd must be non-empty");
        }
        command = format!(
            "cd {} && exec /bin/sh -lc {}",
            sh_quote(cwd),
            sh_quote(&command)
        );
    }
    if options.background {
        command = format!(
            "nohup /bin/sh -lc {} </dev/null >/dev/null 2>&1 &",
            sh_quote(&command)
        );
    }
    Ok(vec!["/bin/sh".to_string(), "-lc".to_string(), command])
}

/// Parameters captured at launch and reused on every reconnect.
#[doc(hidden)]
#[derive(Debug, Clone)]
pub struct ExecParams {
    /// The sailbox the command runs in.
    pub sailbox_id: String,
    /// Worker-proxy endpoint that terminates the exec RPCs for this sailbox.
    pub exec_endpoint: String,
    /// The command and its arguments.
    pub argv: Vec<String>,
    /// Wall-clock limit in seconds before the server kills the command; 0 means
    /// no limit.
    pub timeout_seconds: u32,
    /// Stable key that dedupes the launch and identifies the stream so a
    /// reconnect reattaches to the same command rather than starting a new one.
    pub idempotency_key: String,
    /// Whether the command's stdin is left open for writes.
    pub open_stdin: bool,
    /// Whether to allocate a pseudo-terminal for the command.
    pub pty: bool,
    /// TERM value for the pty (e.g. `xterm-256color`); empty when not a pty.
    pub term: String,
    /// Initial pty width in columns.
    pub cols: u32,
    /// Initial pty height in rows.
    pub rows: u32,
    /// Budget in seconds for retrying transient failures while opening or
    /// resuming the stream.
    pub retry_timeout: f64,
    /// Tracing metadata the wrapper injects (e.g. Voyages); opaque to the core.
    pub extra_metadata: Vec<(String, String)>,
}

/// Incremental UTF-8 decoder: emits only complete characters and carries an
/// incomplete trailing sequence to the next chunk, replacing genuinely invalid
/// bytes with U+FFFD.
#[derive(Default)]
struct Utf8Decoder {
    pending: Vec<u8>,
}

impl Utf8Decoder {
    fn decode(&mut self, data: &[u8]) -> String {
        self.pending.extend_from_slice(data);
        let mut out = String::new();
        loop {
            match std::str::from_utf8(&self.pending) {
                Ok(valid) => {
                    out.push_str(valid);
                    self.pending.clear();
                    break;
                }
                Err(err) => {
                    let valid_up_to = err.valid_up_to();
                    // bytes [..valid_up_to] are valid UTF-8 by definition, so this can't fail.
                    out.push_str(std::str::from_utf8(&self.pending[..valid_up_to]).unwrap());
                    if let Some(bad) = err.error_len() {
                        out.push('\u{FFFD}');
                        self.pending.drain(..valid_up_to + bad);
                    } else {
                        // An incomplete trailing char: keep it for next time.
                        self.pending.drain(..valid_up_to);
                        break;
                    }
                }
            }
        }
        out
    }

    fn flush(&mut self) -> String {
        if self.pending.is_empty() {
            return String::new();
        }
        let out = String::from_utf8_lossy(&self.pending).into_owned();
        self.pending.clear();
        out
    }
}

/// Drop-oldest UTF-8 ring mirroring the server output ring. Appends never
/// block; past the cap the oldest bytes are dropped and `dropped` latches.
/// Pieces carry absolute indices so a reader that falls behind skips the
/// dropped head instead of stalling.
#[derive(Default)]
struct Ring {
    pieces: Vec<String>,
    piece_bytes: Vec<usize>,
    first_idx: usize,
    size: usize,
    dropped: bool,
}

impl Ring {
    fn append(&mut self, text: String) {
        let bytes = text.len();
        self.pieces.push(text);
        self.piece_bytes.push(bytes);
        self.size += bytes;
        while self.size > STREAM_BUFFER_CAP_BYTES {
            let overflow = self.size - STREAM_BUFFER_CAP_BYTES;
            if self.piece_bytes[0] <= overflow {
                self.pieces.remove(0);
                self.size -= self.piece_bytes.remove(0);
                self.first_idx += 1;
            } else {
                // Clip the oldest piece by the overflow. Advance the cut to the
                // next char boundary so a split multibyte char is dropped, then
                // slice the valid str directly: from_utf8_lossy would expand a
                // split leading byte into U+FFFD (3 bytes), which can keep size
                // above the cap and spin this loop while holding the mutex.
                let piece = &self.pieces[0];
                let mut cut = overflow;
                while cut < piece.len() && !piece.is_char_boundary(cut) {
                    cut += 1;
                }
                let kept = piece[cut..].to_string();
                self.size -= self.piece_bytes[0];
                self.piece_bytes[0] = kept.len();
                self.size += self.piece_bytes[0];
                self.pieces[0] = kept;
            }
            self.dropped = true;
        }
    }

    fn tail(&self) -> String {
        self.pieces.concat()
    }
}

#[derive(Default)]
struct State {
    stdout: Ring,
    stderr: Ring,
    ended: bool,
}

impl State {
    fn ring(&self, which: OutputStream) -> &Ring {
        match which {
            OutputStream::Stdout => &self.stdout,
            OutputStream::Stderr => &self.stderr,
        }
    }
}

/// Terminal exec result captured from the Exit frame or a poll.
#[derive(Clone)]
struct ExitInfo {
    status: i32,
    exit_code: i32,
    timed_out: bool,
    stdout_truncated: bool,
    stderr_truncated: bool,
    error_message: String,
    stdout_seq: i64,
    stderr_seq: i64,
}

#[derive(Default)]
struct StdinState {
    offset: i64,
    eof_sent: bool,
    broken: bool,
    /// Set under the lock for the duration of a data write, which holds the lock
    /// across its network send. A clean return clears it; a write whose future
    /// is dropped mid-send (the caller cancelled it) releases the lock with this
    /// still set, so the next writer observes it and poisons rather than
    /// resuming from a stale offset. This is the cancellation latch: it lives in
    /// the same lock that serializes writes, so no later write can race ahead of
    /// it.
    write_in_flight: bool,
}

struct ExecShared {
    worker: Arc<WorkerProxy>,
    params: ExecParams,
    state: Mutex<State>,
    /// Wakes synchronous readers/waiters when output is appended or the stream
    /// ends.
    cond: Condvar,
    /// The async counterpart of `cond`: wakes [`AsyncStreamReader`]s without
    /// parking a runtime thread. Notified on every append and at end of stream.
    data_notify: Notify,
    exit: Mutex<Option<ExitInfo>>,
    /// Highest chunk seq received per stream, published when the pump ends.
    high_seq: Mutex<(i64, i64)>,
    stdin: AsyncMutex<StdinState>,
    ended: AtomicBool,
    ended_notify: Notify,
    closing: AtomicBool,
    close_notify: Notify,
}

/// A handle to a running command. Drop or [`ExecProcess::close`] releases the
/// stream without killing the command.
pub struct ExecProcess {
    shared: Arc<ExecShared>,
    exec_request_id: String,
}

impl Drop for ExecProcess {
    fn drop(&mut self) {
        // Honor the documented contract: a handle dropped without an explicit
        // close() (e.g. by Python GC) still stops the pump and releases the
        // stream, instead of leaving the gRPC stream running until the command
        // finishes on its own.
        self.close();
    }
}

impl ExecProcess {
    /// Submit the exec and start pumping output. Blocks until the server sends
    /// the `Started` frame (the launch is durably settled).
    ///
    /// # Runtime
    ///
    /// Spawns the background output pump on the calling task's tokio runtime, so
    /// call it from within one. The reconnect dials co-locate on that runtime.
    #[doc(hidden)]
    pub async fn start(
        worker: Arc<WorkerProxy>,
        mut params: ExecParams,
    ) -> Result<ExecProcess, SailError> {
        // The idempotency key dedupes the launch and lets a reconnect reattach
        // to the same command. Mint one when the caller didn't supply their own,
        // so every binding gets a stable key without restating the format.
        let key = params.idempotency_key.trim();
        params.idempotency_key = if key.is_empty() {
            format!("exec_{}", uuid::Uuid::new_v4())
        } else {
            key.to_string()
        };
        let (exec_request_id, stream) = submit(&worker, &params, 0, 0).await?;
        let shared = Arc::new(ExecShared {
            worker,
            params,
            state: Mutex::new(State::default()),
            cond: Condvar::new(),
            data_notify: Notify::new(),
            exit: Mutex::new(None),
            high_seq: Mutex::new((0, 0)),
            stdin: AsyncMutex::new(StdinState::default()),
            ended: AtomicBool::new(false),
            ended_notify: Notify::new(),
            closing: AtomicBool::new(false),
            close_notify: Notify::new(),
        });
        let pump_shared = shared.clone();
        tokio::spawn(async move { pump(pump_shared, stream).await });
        Ok(ExecProcess {
            shared,
            exec_request_id,
        })
    }

    /// The durable server-assigned id for this exec, taken from the `Started`
    /// frame. Identifies the command for wait, cancel, resize, and stdin RPCs.
    pub fn exec_request_id(&self) -> &str {
        &self.exec_request_id
    }

    /// The idempotency key this exec launched with: the caller's, or the one
    /// minted in `start` when none was supplied.
    pub fn idempotency_key(&self) -> &str {
        &self.shared.params.idempotency_key
    }

    /// Create a reader over a live output stream. A fresh reader replays the
    /// retained tail from the start, then follows live.
    pub fn reader(&self, which: OutputStream) -> StreamReader {
        StreamReader {
            shared: self.shared.clone(),
            which,
            cursor: 0,
        }
    }

    /// Create an async reader over a live output stream (the awaiting twin of
    /// [`reader`](Self::reader)).
    pub fn reader_async(&self, which: OutputStream) -> AsyncStreamReader {
        AsyncStreamReader {
            shared: self.shared.clone(),
            which,
            cursor: 0,
        }
    }

    /// Returns the exit code if the Exit frame arrived on the stream, mapping a
    /// not-a-real-result terminal status to its error. `None` means the result
    /// is not known on the stream yet. `wait` is authoritative.
    pub fn poll(&self) -> Option<Result<i32, SailError>> {
        let exit = lock(&self.shared.exit);
        exit.as_ref()
            .map(|exit| match terminal_status_error(exit.status) {
                Some(err) => Err(err),
                None => Ok(exit.exit_code),
            })
    }

    /// Stop the pump and release the stream without touching the remote command.
    pub fn close(&self) {
        self.shared.closing.store(true, Ordering::SeqCst);
        // notify_one stores a permit if the pump is between its `closing` check
        // and registering on close_notify, so a close racing the pump is not
        // missed (notify_waiters wakes only already-registered waiters). The
        // pump is the sole waiter, so one permit suffices.
        self.shared.close_notify.notify_one();
    }

    /// Block up to `timeout` for the output stream to end; returns whether it
    /// has. Lets a synchronous caller stay responsive to its own signals and
    /// stop conditions between ticks before committing to the blocking
    /// [`wait`](Self::wait) resolve.
    pub async fn wait_stream_ended(&self, timeout: Duration) -> bool {
        let _ = tokio::time::timeout(timeout, self.await_ended()).await;
        self.shared.ended.load(Ordering::SeqCst)
    }

    /// Resolve once the output stream has ended (the pump set `ended`).
    async fn await_ended(&self) {
        loop {
            let notified = self.shared.ended_notify.notified();
            if self.shared.ended.load(Ordering::SeqCst) {
                return;
            }
            notified.await;
        }
    }

    /// Wait for the command to finish and return its buffered result.
    ///
    /// A clean Exit resolves from the local rings; a stream that ended without
    /// one (or whose tail is truncated or short) falls back to `WaitSailboxExec`,
    /// which reattaches to the guest session for the authoritative result,
    /// retrying a not-ready box for the exec's configured `retry_timeout` budget.
    pub async fn wait(&self) -> Result<ExecResult, SailError> {
        self.await_ended().await;
        let exit = lock(&self.shared.exit).clone();
        let (high_out, high_err) = *lock(&self.shared.high_seq);
        let (out_dropped, err_dropped) = {
            let state = lock(&self.shared.state);
            (state.stdout.dropped, state.stderr.dropped)
        };

        // The local rings are authoritative only for a clean, untruncated, fully
        // delivered stream. Otherwise fall back to WaitSailboxExec, which
        // reattaches to the guest.
        let truncated = exit.as_ref().is_some_and(|exit| {
            exit.stdout_truncated || exit.stderr_truncated || out_dropped || err_dropped
        });
        let incomplete = exit
            .as_ref()
            .is_some_and(|exit| exit.stdout_seq > high_out || exit.stderr_seq > high_err);

        // Whether the live stream reached each stream's final chunk. When true,
        // a live consumer already saw the whole stream (the buffered tail below
        // can only repeat its ending); when false, the buffered copy is the only
        // way to recover the missing tail.
        let stdout_complete = exit
            .as_ref()
            .is_some_and(|exit| exit.stdout_seq <= high_out);
        let stderr_complete = exit
            .as_ref()
            .is_some_and(|exit| exit.stderr_seq <= high_err);

        if exit.is_none() || truncated || incomplete {
            let outcome = self
                .shared
                .worker
                .wait_exec(
                    &self.shared.params.exec_endpoint,
                    &self.shared.params.sailbox_id,
                    &self.exec_request_id,
                    self.shared.params.retry_timeout,
                )
                .await?;
            // Record the polled terminal outcome (when the stream carried no
            // Exit) so poll()/returncode agree with this wait().
            {
                let mut exit_slot = lock(&self.shared.exit);
                if exit_slot.is_none() {
                    *exit_slot = Some(ExitInfo {
                        status: outcome.status,
                        exit_code: outcome.exit_code,
                        timed_out: outcome.timed_out,
                        stdout_truncated: outcome.stdout_truncated,
                        stderr_truncated: outcome.stderr_truncated,
                        error_message: String::new(),
                        stdout_seq: 0,
                        stderr_seq: 0,
                    });
                }
            }
            // A real terminal Exit was already witnessed on the live stream, so
            // a worker_lost poll is stale: the worker can be lost between the
            // command finishing and the persisted row being read. The witnessed
            // completion is authoritative, so return it from the local rings
            // rather than raising worker-lost.
            if let Some(witnessed) = exit.as_ref() {
                if outcome.status == pb::SailboxExecStatus::WorkerLost as i32 {
                    let state = lock(&self.shared.state);
                    let mut stderr = state.stderr.tail();
                    if witnessed.status == pb::SailboxExecStatus::Failed as i32
                        && !witnessed.error_message.is_empty()
                    {
                        stderr = witnessed.error_message.clone();
                    }
                    return Ok(ExecResult {
                        stdout: state.stdout.tail(),
                        stderr,
                        exit_code: witnessed.exit_code,
                        timed_out: witnessed.timed_out,
                        stdout_truncated: witnessed.stdout_truncated
                            || out_dropped
                            || witnessed.stdout_seq > high_out,
                        stderr_truncated: witnessed.stderr_truncated
                            || err_dropped
                            || witnessed.stderr_seq > high_err,
                        stdout_complete,
                        stderr_complete,
                    });
                }
            }
            if let Some(err) = terminal_status_error(outcome.status) {
                return Err(err);
            }
            return Ok(ExecResult {
                stdout: outcome.stdout,
                stderr: outcome.stderr,
                exit_code: outcome.exit_code,
                timed_out: outcome.timed_out,
                stdout_truncated: outcome.stdout_truncated,
                stderr_truncated: outcome.stderr_truncated,
                stdout_complete,
                stderr_complete,
            });
        }

        let exit = exit.expect("exit present on the clean path");
        if let Some(err) = terminal_status_error(exit.status) {
            return Err(err);
        }
        let state = lock(&self.shared.state);
        let mut stderr = state.stderr.tail();
        if exit.status == pb::SailboxExecStatus::Failed as i32 && !exit.error_message.is_empty() {
            // A failed row persists its failure text as stderr; mirror the poll path.
            stderr = exit.error_message.clone();
        }
        Ok(ExecResult {
            stdout: state.stdout.tail(),
            stderr,
            exit_code: exit.exit_code,
            timed_out: exit.timed_out,
            stdout_truncated: false,
            stderr_truncated: false,
            stdout_complete,
            stderr_complete,
        })
    }

    /// Signal the command: [`CancelSignal::Interrupt`] (SIGINT) or
    /// [`CancelSignal::Kill`] (SIGKILL).
    pub async fn cancel(&self, signal: CancelSignal, retry: RetryBudget) -> Result<(), SailError> {
        self.shared
            .worker
            .cancel_exec(
                &self.shared.params.exec_endpoint,
                &self.shared.params.sailbox_id,
                &self.exec_request_id,
                signal.is_force(),
                retry.as_secs_f64(),
            )
            .await
    }

    /// Set the pty window for a `pty` exec. Advisory and best-effort: an
    /// unknown, finished, or not-yet-placed exec is a server no-op, and a
    /// transient transport error is swallowed (the next resize resends).
    pub async fn resize(&self, cols: u32, rows: u32) {
        let message = pb::ResizeSailboxExecRequest {
            sailbox_id: self.shared.params.sailbox_id.clone(),
            exec_request_id: self.exec_request_id.clone(),
            cols,
            rows,
        };
        let Ok(request) =
            self.shared
                .worker
                .request_for(message, &[], Some(Duration::from_secs(5)))
        else {
            return;
        };
        if let Ok(mut client) = self
            .shared
            .worker
            .client_for(&self.shared.params.exec_endpoint)
        {
            let _ = client.resize_sailbox_exec(request).await;
        }
    }

    /// Write to the command's stdin. Chunked with absolute offsets; an uncertain
    /// mid-flight failure poisons the writer (a stale-offset resume could
    /// silently drop bytes). Blocks (with backoff) while the guest buffer is full.
    pub async fn write_stdin(&self, data: &[u8]) -> Result<(), SailError> {
        let mut stdin = self.shared.stdin.lock().await;
        if stdin.eof_sent {
            return Err(SailError::BrokenPipe {
                message: "stdin is closed".to_string(),
            });
        }
        if stdin.broken {
            return Err(SailError::BrokenPipe {
                message: "an earlier stdin write failed".to_string(),
            });
        }
        if stdin.write_in_flight {
            // The previous write held the lock across its send and never cleared
            // this, so its future was cancelled mid-flight: bytes may have landed
            // and the offset is uncertain. Poison rather than resume.
            stdin.broken = true;
            return Err(SailError::BrokenPipe {
                message: "an earlier stdin write was interrupted".to_string(),
            });
        }
        if data.is_empty() {
            return Ok(());
        }
        stdin.write_in_flight = true;
        let result = self.send_stdin(&mut stdin, data, false).await;
        // Reached only if the send was not cancelled. A clean return clears the
        // latch; an uncertain failure already set `broken` inside send_stdin.
        stdin.write_in_flight = false;
        result
    }

    /// Close the command's stdin (send EOF).
    pub async fn close_stdin(&self) -> Result<(), SailError> {
        let mut stdin = self.shared.stdin.lock().await;
        if stdin.eof_sent {
            return Ok(());
        }
        if stdin.broken || stdin.write_in_flight {
            // An earlier write failed or was cancelled mid-flight, so the stream
            // is undeliverable at a known offset.
            stdin.eof_sent = true;
            return Ok(());
        }
        // The EOF write carries no data and is idempotent, so a transient
        // failure can't corrupt the offset: send directly without poisoning, and
        // leave eof_sent false until it lands so a later eof retries it.
        match self.send_stdin(&mut stdin, &[], true).await {
            Ok(()) => {
                stdin.eof_sent = true;
                Ok(())
            }
            // The command already exited / closed stdin: nothing to deliver.
            Err(SailError::BrokenPipe { .. }) => {
                stdin.eof_sent = true;
                Ok(())
            }
            Err(err) => Err(err),
        }
    }

    /// Drive the chunked WriteSailboxExecStdin loop. `eof` latches only when the
    /// final chunk is fully accepted. Poisons `stdin.broken` on an uncertain
    /// mid-flight failure (anything but a clean broken-pipe).
    async fn send_stdin(
        &self,
        stdin: &mut StdinState,
        payload: &[u8],
        eof: bool,
    ) -> Result<(), SailError> {
        let endpoint = &self.shared.params.exec_endpoint;
        let sailbox_id = &self.shared.params.sailbox_id;
        let mut deadline = retry_deadline(EXEC_TRANSIENT_RETRY_TIMEOUT_SECONDS);
        let mut delay = EXEC_TRANSIENT_RETRY_INITIAL_DELAY_SECONDS;
        let mut sent = 0usize;
        loop {
            let end = (sent + STDIN_WRITE_CHUNK_BYTES).min(payload.len());
            let chunk = &payload[sent..end];
            let last = end >= payload.len();
            let message = pb::WriteSailboxExecStdinRequest {
                sailbox_id: sailbox_id.clone(),
                exec_request_id: self.exec_request_id.clone(),
                offset: stdin.offset,
                data: chunk.to_vec(),
                eof: eof && last,
            };
            // Bound each attempt so a stalled connection (one that never
            // returns a status) times out and the retry/poison logic below
            // runs, instead of one await blocking the whole budget.
            let request = match self.shared.worker.request_for(
                message,
                &[],
                Some(rpc_attempt_timeout(deadline)),
            ) {
                Ok(request) => request,
                Err(err) => return Err(err),
            };
            let result = match self.shared.worker.client_for(endpoint) {
                Ok(mut client) => client.write_sailbox_exec_stdin(request).await,
                Err(err) => return Err(err),
            };
            match result {
                Ok(resp) => {
                    let accepted_through = resp.into_inner().accepted_through;
                    // Clamp to the chunk we actually sent: a server that
                    // over-reports accepted_through must not push `sent` past the
                    // payload (a slice panic) or advance the idempotent offset
                    // beyond delivered bytes.
                    let accepted =
                        ((accepted_through - stdin.offset).max(0) as usize).min(chunk.len());
                    stdin.offset += accepted as i64;
                    sent += accepted;
                    if sent >= payload.len() && (!eof || (last && accepted == chunk.len())) {
                        return Ok(());
                    }
                    // A success proves the transport healthy: restart the budget.
                    deadline = retry_deadline(EXEC_TRANSIENT_RETRY_TIMEOUT_SECONDS);
                    if accepted > 0 {
                        delay = EXEC_TRANSIENT_RETRY_INITIAL_DELAY_SECONDS;
                    } else {
                        // Guest buffer full: block like a pipe write, no deadline.
                        delay = sleep_no_deadline(delay).await;
                    }
                }
                Err(status) => {
                    if matches!(status.code(), Code::NotFound | Code::FailedPrecondition) {
                        // The exec is over or closed its stdin: a dead pipe.
                        return Err(SailError::BrokenPipe {
                            message: status.message().to_string(),
                        });
                    }
                    if is_exec_not_ready(&status) {
                        // Row open but guest not reachable yet (wake/migration):
                        // wait it out against the exec's lifetime, no deadline,
                        // resetting the transport budget on each "; retry".
                        delay = sleep_no_deadline(delay).await;
                        deadline = retry_deadline(EXEC_TRANSIENT_RETRY_TIMEOUT_SECONDS);
                        continue;
                    }
                    if !should_retry_transient_exec_rpc(&status, deadline) {
                        // Uncertain whether bytes landed: poison so a stale-offset
                        // resume can't silently drop overlapping bytes.
                        stdin.broken = true;
                        return Err(SailError::from_exec_status(&status));
                    }
                    tracing::warn!(code = ?status.code(), "retrying exec stdin write");
                    if should_invalidate_channel(&status) {
                        self.shared.worker.channels().invalidate(endpoint);
                    }
                    delay = sleep_before_retry(delay, deadline).await;
                }
            }
        }
    }
}

/// A cursor over one live output stream. [`StreamReader::next`] blocks up to a
/// timeout for the next chunk.
pub struct StreamReader {
    shared: Arc<ExecShared>,
    which: OutputStream,
    cursor: usize,
}

impl StreamReader {
    /// Block up to `timeout` for the next step: the next retained chunk, `Eof`
    /// once the stream is closed and fully drained, or `Pending` if nothing new
    /// arrived in time.
    pub fn next(&mut self, timeout: Duration) -> ReadStep {
        let mut state = lock(&self.shared.state);
        loop {
            let ring = state.ring(self.which);
            if self.cursor < ring.first_idx {
                self.cursor = ring.first_idx;
            }
            let available = ring.first_idx + ring.pieces.len();
            if self.cursor < available {
                let piece = ring.pieces[self.cursor - ring.first_idx].clone();
                self.cursor += 1;
                return ReadStep::Chunk(piece);
            }
            if state.ended {
                return ReadStep::Eof;
            }
            let (next_state, timed_out) = self
                .shared
                .cond
                .wait_timeout(state, timeout)
                .expect("exec state mutex poisoned");
            state = next_state;
            if timed_out.timed_out() {
                return ReadStep::Pending;
            }
        }
    }
}

/// An async cursor over one live output stream, the awaiting counterpart of
/// [`StreamReader`]. [`AsyncStreamReader::next`] yields the next chunk without
/// parking a runtime thread, so many streams can be read on one event loop.
pub struct AsyncStreamReader {
    shared: Arc<ExecShared>,
    which: OutputStream,
    cursor: usize,
}

impl AsyncStreamReader {
    /// The next retained chunk, or `None` once the stream is closed and fully
    /// drained. A reader that falls more than the buffer cap behind skips the
    /// dropped head rather than stalling.
    pub async fn next(&mut self) -> Option<String> {
        loop {
            // Arm the wakeup before inspecting the ring: `notify_waiters` only
            // wakes already-registered waiters, so enabling first closes the gap
            // where an append between the check and the await would be missed.
            let notified = self.shared.data_notify.notified();
            tokio::pin!(notified);
            notified.as_mut().enable();
            {
                let state = lock(&self.shared.state);
                let ring = state.ring(self.which);
                if self.cursor < ring.first_idx {
                    self.cursor = ring.first_idx;
                }
                let available = ring.first_idx + ring.pieces.len();
                if self.cursor < available {
                    let piece = ring.pieces[self.cursor - ring.first_idx].clone();
                    self.cursor += 1;
                    return Some(piece);
                }
                if state.ended {
                    return None;
                }
            }
            notified.await;
        }
    }
}

/// Sleep one backoff step with no deadline (used when the guest stdin buffer is
/// full or the guest is not reachable yet); returns the doubled delay.
async fn sleep_no_deadline(delay: f64) -> f64 {
    let sleep_for = delay.min(crate::worker::EXEC_TRANSIENT_RETRY_MAX_DELAY_SECONDS);
    tokio::time::sleep(Duration::from_secs_f64(sleep_for.max(0.0))).await;
    (delay * 2.0).min(crate::worker::EXEC_TRANSIENT_RETRY_MAX_DELAY_SECONDS)
}

fn is_exec_not_ready(status: &Status) -> bool {
    status.code() == Code::Unavailable && status.message().contains("; retry")
}

/// Map a worker-lost status (no real return code) to its error; `None` for a
/// normal exit (succeeded/failed/timed-out, which carry a real return code).
fn terminal_status_error(status: i32) -> Option<SailError> {
    if status == pb::SailboxExecStatus::WorkerLost as i32 {
        return Some(SailError::WorkerLost {
            message: "the machine hosting your sailbox was lost before the command \
                      finished; run exec again to retry"
                .to_string(),
        });
    }
    None
}

/// Open the exec stream and read the `Started` frame, retrying transient
/// failures. Non-zero resume seqs reattach a dropped stream.
async fn submit(
    worker: &Arc<WorkerProxy>,
    params: &ExecParams,
    stdout_resume_seq: i64,
    stderr_resume_seq: i64,
) -> Result<(String, Streaming<pb::StreamSailboxExecResponse>), SailError> {
    let deadline = retry_deadline(params.retry_timeout);
    let mut delay = EXEC_TRANSIENT_RETRY_INITIAL_DELAY_SECONDS;
    loop {
        let message = pb::StreamSailboxExecRequest {
            sailbox_id: params.sailbox_id.clone(),
            argv: params.argv.clone(),
            timeout_seconds: params.timeout_seconds,
            idempotency_key: params.idempotency_key.clone(),
            open_stdin: params.open_stdin,
            stdout_resume_seq,
            stderr_resume_seq,
            pty: params.pty,
            term_cols: params.cols,
            term_rows: params.rows,
            term: params.term.clone(),
        };
        let request = worker.request_for(message, &params.extra_metadata, None)?;
        let status = match worker
            .client_for(&params.exec_endpoint)?
            .stream_sailbox_exec(request)
            .await
        {
            Ok(resp) => {
                let mut stream = resp.into_inner();
                match stream.message().await {
                    Ok(Some(first)) => match first.frame {
                        Some(pb::stream_sailbox_exec_response::Frame::Started(started)) => {
                            return Ok((started.exec_request_id, stream));
                        }
                        _ => {
                            return Err(SailError::Execution {
                                code: crate::error::GrpcCode::Internal,
                                detail: "exec stream opened with a non-started frame".to_string(),
                            });
                        }
                    },
                    // On a fresh launch (resume seqs 0,0) a clean end before the
                    // Started frame is a transport-level teardown, not a server
                    // verdict — the server deliberately sends Started even for
                    // lost sessions on that path. Nothing was confirmed and the
                    // relaunch reuses the idempotency key, so route it through
                    // the transient-retry gate like any dropped connection. On a
                    // mid-run reconnect the same clean end IS a verdict (the box
                    // parked: autoslept, lost, or re-homing) — surface it so the
                    // caller falls back to wait(), which wakes the box, instead
                    // of re-submitting against a server that will not act.
                    Ok(None) if stdout_resume_seq == 0 && stderr_resume_seq == 0 => {
                        tonic::Status::unavailable(
                            "exec stream ended before the server confirmed the launch",
                        )
                    }
                    Ok(None) => {
                        return Err(SailError::Execution {
                            code: crate::error::GrpcCode::Internal,
                            detail: "exec stream ended before the server confirmed the launch"
                                .to_string(),
                        });
                    }
                    Err(status) => status,
                }
            }
            Err(status) => status,
        };
        if !should_retry_transient_exec_rpc(&status, deadline) {
            return Err(SailError::from_exec_status(&status));
        }
        tracing::warn!(
            code = ?status.code(),
            stdout_resume_seq,
            stderr_resume_seq,
            "reconnecting exec stream"
        );
        if should_invalidate_channel(&status) {
            worker.channels().invalidate(&params.exec_endpoint);
        }
        delay = sleep_before_retry(delay, deadline).await;
    }
}

/// Drains exec frames into the rings while tracking the per-stream high-water
/// seqs (the resume points sent on reconnect) and carrying the incremental UTF-8
/// decoders across reconnects. Split out from the transport loop so the
/// resume/replay/decode state machine is testable in-process without a gRPC
/// stream: a mid-stream break is just "keep applying frames after a gap".
struct Pump {
    shared: Arc<ExecShared>,
    stdout_dec: Utf8Decoder,
    stderr_dec: Utf8Decoder,
    stdout_seq: i64,
    stderr_seq: i64,
}

impl Pump {
    fn new(shared: Arc<ExecShared>) -> Pump {
        Pump {
            shared,
            stdout_dec: Utf8Decoder::default(),
            stderr_dec: Utf8Decoder::default(),
            stdout_seq: 0,
            stderr_seq: 0,
        }
    }

    /// Apply one frame. Returns `true` for a terminal Exit frame (stop draining).
    /// Chunk seqs only advance the high-water mark, never rewind, so a server
    /// that replays an already-seen tail after reconnect can't lower the resume
    /// point.
    fn apply_frame(&mut self, frame: pb::StreamSailboxExecResponse) -> bool {
        match frame.frame {
            Some(pb::stream_sailbox_exec_response::Frame::Chunk(chunk)) => {
                let is_stderr = chunk.stream == pb::SailboxExecStream::Stderr as i32;
                let (decoder, which) = if is_stderr {
                    self.stderr_seq = self.stderr_seq.max(chunk.seq);
                    (&mut self.stderr_dec, OutputStream::Stderr)
                } else {
                    self.stdout_seq = self.stdout_seq.max(chunk.seq);
                    (&mut self.stdout_dec, OutputStream::Stdout)
                };
                let text = decoder.decode(&chunk.data);
                if !text.is_empty() {
                    let mut state = lock(&self.shared.state);
                    match which {
                        OutputStream::Stdout => state.stdout.append(text),
                        OutputStream::Stderr => state.stderr.append(text),
                    }
                    self.shared.cond.notify_all();
                    self.shared.data_notify.notify_waiters();
                }
                false
            }
            Some(pb::stream_sailbox_exec_response::Frame::Exit(exit)) => {
                *lock(&self.shared.exit) = Some(ExitInfo {
                    status: exit.status,
                    exit_code: exit.return_code,
                    timed_out: exit.timed_out,
                    stdout_truncated: exit.stdout_truncated,
                    stderr_truncated: exit.stderr_truncated,
                    error_message: exit.error_message,
                    stdout_seq: exit.stdout_seq,
                    stderr_seq: exit.stderr_seq,
                });
                true
            }
            _ => false,
        }
    }

    /// Flush the decoders, publish the high-water seqs, close the rings, and wake
    /// every reader and waiter. Consumes the pump: nothing follows finalize.
    fn finalize(mut self) {
        let tail_out = self.stdout_dec.flush();
        let tail_err = self.stderr_dec.flush();
        {
            let mut state = lock(&self.shared.state);
            if !tail_out.is_empty() {
                state.stdout.append(tail_out);
            }
            if !tail_err.is_empty() {
                state.stderr.append(tail_err);
            }
            state.ended = true;
            self.shared.cond.notify_all();
            self.shared.data_notify.notify_waiters();
        }
        *lock(&self.shared.high_seq) = (self.stdout_seq, self.stderr_seq);
        self.shared.ended.store(true, Ordering::SeqCst);
        self.shared.ended_notify.notify_waiters();
    }
}

/// Drain the output stream into the rings, reconnecting on a transient break,
/// until the Exit frame or an unrecoverable end. Always finalizes the rings.
async fn pump(shared: Arc<ExecShared>, mut stream: Streaming<pb::StreamSailboxExecResponse>) {
    let mut state = Pump::new(shared.clone());
    loop {
        if shared.closing.load(Ordering::SeqCst) {
            break;
        }
        let message = tokio::select! {
            biased;
            () = shared.close_notify.notified() => break,
            message = stream.message() => message,
        };
        match message {
            Ok(Some(frame)) => {
                if state.apply_frame(frame) {
                    break;
                }
            }
            // The stream ended cleanly without an Exit; leave it to wait()'s poll.
            Ok(None) => break,
            Err(_status) => {
                if shared.closing.load(Ordering::SeqCst) {
                    break;
                }
                // Reconnect from the last seq we saw, so the guest replays only
                // the unseen tail.
                match submit(
                    &shared.worker,
                    &shared.params,
                    state.stdout_seq,
                    state.stderr_seq,
                )
                .await
                {
                    Ok((_id, fresh)) => {
                        stream = fresh;
                        continue;
                    }
                    Err(_) => break,
                }
            }
        }
    }
    state.finalize();
}

/// Fuzz entry point: drive an arbitrary byte stream through the incremental
/// UTF-8 decoder (split at data-derived boundaries) and the drop-oldest ring,
/// asserting the engine's invariants. Any violation panics, which libfuzzer
/// flags as a crash. Compiled only under `cfg(test)` or the `fuzzing` feature,
/// so it is never part of a production build.
#[cfg(any(test, feature = "fuzzing"))]
pub fn fuzz_decode_ring(data: &[u8]) {
    // Decoding the whole input at once.
    let mut whole = Utf8Decoder::default();
    let mut whole_out = whole.decode(data);
    whole_out.push_str(&whole.flush());

    // Decoding the same bytes split at boundaries derived from the data itself
    // (cut after every odd byte), to exercise mid-char breaks.
    let mut chunked = Utf8Decoder::default();
    let mut chunked_out = String::new();
    let mut start = 0;
    for (i, byte) in data.iter().enumerate() {
        if byte & 1 == 1 {
            chunked_out.push_str(&chunked.decode(&data[start..=i]));
            start = i + 1;
        }
    }
    chunked_out.push_str(&chunked.decode(&data[start..]));
    chunked_out.push_str(&chunked.flush());

    // The decoder is chunk-boundary invariant and agrees with a lossy decode.
    assert_eq!(
        chunked_out, whole_out,
        "decoder is not chunk-boundary invariant"
    );
    assert_eq!(
        whole_out,
        String::from_utf8_lossy(data),
        "decoder disagrees with a lossy decode of the whole input"
    );

    // Push the decoded fuzz output through the ring, then overrun the cap by a
    // hair so the drop-oldest clip lands inside the fuzz-derived front piece:
    // the ring must stay within the cap and never split a char (no panic), and
    // its retained bytes must remain a suffix of everything appended.
    let mut ring = Ring::default();
    ring.append(whole_out.clone());
    let filler = STREAM_BUFFER_CAP_BYTES + 1 - whole_out.len().min(STREAM_BUFFER_CAP_BYTES);
    let filler = "a".repeat(filler);
    ring.append(filler.clone());
    assert!(
        ring.size <= STREAM_BUFFER_CAP_BYTES,
        "ring exceeded its cap"
    );
    let full = format!("{whole_out}{filler}");
    let tail = ring.tail();
    assert!(
        full.as_bytes().ends_with(tail.as_bytes()),
        "ring tail is not a suffix of the appended bytes"
    );
}

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

    #[test]
    fn shell_argv_wraps_cwd_and_background() {
        let plain = shell_argv("echo hi", &ExecOptions::default()).unwrap();
        assert_eq!(plain, ["/bin/sh", "-lc", "echo hi"]);

        let cwd = shell_argv(
            "echo hi",
            &ExecOptions {
                cwd: Some("/app".to_string()),
                ..ExecOptions::default()
            },
        )
        .unwrap();
        assert_eq!(cwd[2], "cd '/app' && exec /bin/sh -lc 'echo hi'");

        let background = shell_argv(
            "echo hi",
            &ExecOptions {
                cwd: Some("/app".to_string()),
                background: true,
                ..ExecOptions::default()
            },
        )
        .unwrap();
        assert_eq!(
            background[2],
            "nohup /bin/sh -lc 'cd '\\''/app'\\'' && exec /bin/sh -lc '\\''echo hi'\\''' </dev/null >/dev/null 2>&1 &"
        );
    }

    #[test]
    fn shell_argv_rejects_invalid_combinations() {
        assert!(shell_argv("", &ExecOptions::default()).is_err());
        assert!(shell_argv(
            "x",
            &ExecOptions {
                cwd: Some("   ".to_string()),
                ..ExecOptions::default()
            }
        )
        .is_err());
        for (open_stdin, pty) in [(true, false), (false, true)] {
            assert!(shell_argv(
                "x",
                &ExecOptions {
                    background: true,
                    open_stdin,
                    pty,
                    ..ExecOptions::default()
                }
            )
            .is_err());
        }
    }

    #[test]
    fn fuzz_decode_ring_holds_on_samples() {
        // Verifies the fuzz entry point itself: hand-picked inputs covering valid
        // multibyte chars, split sequences, and invalid bytes.
        for sample in [
            &b""[..],
            b"hello",
            b"\xff\xfe\xfd",
            "€ µ é".as_bytes(),
            b"ab\xc3\xa9cd",
            &[0xC3, 0x28],
        ] {
            fuzz_decode_ring(sample);
        }
    }

    #[test]
    fn ring_drops_oldest_past_cap_and_latches() {
        let mut ring = Ring::default();
        ring.append("a".repeat(STREAM_BUFFER_CAP_BYTES));
        assert!(!ring.dropped);
        ring.append("bbbb".to_string());
        assert!(ring.dropped);
        assert_eq!(ring.tail().len(), STREAM_BUFFER_CAP_BYTES);
        assert!(ring.tail().ends_with("bbbb"));
    }

    #[test]
    fn ring_clip_on_split_multibyte_char_terminates() {
        // Overflow of 1 lands inside the leading 2-byte 'é'. The clip must drop
        // the split char and shrink the buffer; expanding it via lossy
        // replacement could keep size above the cap and spin the trim loop while
        // holding the state mutex. Reaching the asserts at all proves it ends.
        let mut ring = Ring::default();
        ring.append(format!("é{}", "a".repeat(STREAM_BUFFER_CAP_BYTES - 1)));
        assert!(ring.size <= STREAM_BUFFER_CAP_BYTES);
        assert!(ring.dropped);
        let tail = ring.tail();
        assert!(!tail.contains('é'));
        assert!(!tail.contains('\u{FFFD}'));
    }

    #[test]
    fn decoder_carries_split_multibyte_char() {
        let mut dec = Utf8Decoder::default();
        let euro = "€".as_bytes(); // 3 bytes
        assert_eq!(dec.decode(&euro[..2]), "");
        assert_eq!(dec.decode(&euro[2..]), "€");
    }

    #[test]
    fn decoder_replaces_invalid_bytes() {
        let mut dec = Utf8Decoder::default();
        assert_eq!(dec.decode(&[0xff, b'x']), "\u{FFFD}x");
    }

    #[test]
    fn terminal_status_mapping() {
        assert!(matches!(
            terminal_status_error(pb::SailboxExecStatus::WorkerLost as i32),
            Some(SailError::WorkerLost { .. })
        ));
        assert!(terminal_status_error(pb::SailboxExecStatus::Succeeded as i32).is_none());
        // Retired wire values (canceled=9, interrupted_retryable=6, interrupted_unsafe_to_retry=7,
        // reserved in the proto) carry no error: an old backend that still emits one falls through
        // to a normal result rather than raising.
        for retired in [9, 6, 7] {
            assert!(terminal_status_error(retired).is_none());
        }
    }

    fn test_shared() -> Arc<ExecShared> {
        Arc::new(ExecShared {
            worker: Arc::new(WorkerProxy::new("test-key").unwrap()),
            params: ExecParams {
                sailbox_id: "sb".into(),
                exec_endpoint: "endpoint".into(),
                argv: vec!["echo".into()],
                timeout_seconds: 0,
                idempotency_key: "idem".into(),
                open_stdin: false,
                pty: false,
                term: String::new(),
                cols: 0,
                rows: 0,
                retry_timeout: 0.0,
                extra_metadata: vec![],
            },
            state: Mutex::new(State::default()),
            cond: Condvar::new(),
            data_notify: Notify::new(),
            exit: Mutex::new(None),
            high_seq: Mutex::new((0, 0)),
            stdin: AsyncMutex::new(StdinState::default()),
            ended: AtomicBool::new(false),
            ended_notify: Notify::new(),
            closing: AtomicBool::new(false),
            close_notify: Notify::new(),
        })
    }

    fn chunk(which: OutputStream, seq: i64, data: &[u8]) -> pb::StreamSailboxExecResponse {
        let stream = match which {
            OutputStream::Stdout => pb::SailboxExecStream::Stdout,
            OutputStream::Stderr => pb::SailboxExecStream::Stderr,
        };
        pb::StreamSailboxExecResponse {
            frame: Some(pb::stream_sailbox_exec_response::Frame::Chunk(
                pb::SailboxExecChunk {
                    stream: stream as i32,
                    data: data.to_vec(),
                    seq,
                },
            )),
        }
    }

    fn exit_frame(status: pb::SailboxExecStatus, stdout_seq: i64) -> pb::StreamSailboxExecResponse {
        pb::StreamSailboxExecResponse {
            frame: Some(pb::stream_sailbox_exec_response::Frame::Exit(
                pb::SailboxExecExit {
                    status: status as i32,
                    return_code: 0,
                    timed_out: false,
                    stdout_truncated: false,
                    stderr_truncated: false,
                    error_message: String::new(),
                    stdout_seq,
                    stderr_seq: 0,
                },
            )),
        }
    }

    /// The resume state machine: a multibyte char split across a mid-stream break
    /// decodes correctly (the decoder persists across reconnect), the high-water
    /// seq only advances (so a replayed tail can't lower the resume point), and
    /// finalize publishes the seqs `wait()` checks for completeness.
    #[tokio::test]
    async fn exec_resume_carries_utf8_and_tracks_seq_across_break() {
        let shared = test_shared();
        let mut pump = Pump::new(shared.clone());

        // Pre-break: seq 1 ends mid-'é' (0xC3 0xA9), delivering only the lead byte.
        assert!(!pump.apply_frame(chunk(OutputStream::Stdout, 1, b"ab\xc3")));
        assert_eq!(shared.state.lock().unwrap().stdout.tail(), "ab");
        // The reconnect would call submit(.., stdout_resume_seq = 1, ..).
        assert_eq!(pump.stdout_seq, 1);

        // The socket breaks; the guest replays only seq > 1. Seq 2 supplies the
        // rest of 'é' plus more, and the persisted decoder completes the char.
        assert!(!pump.apply_frame(chunk(OutputStream::Stdout, 2, b"\xa9cd")));
        assert_eq!(shared.state.lock().unwrap().stdout.tail(), "abécd");
        assert_eq!(pump.stdout_seq, 2);

        // An out-of-order/replayed lower seq must not rewind the resume point.
        assert!(!pump.apply_frame(chunk(OutputStream::Stdout, 1, b"!")));
        assert_eq!(pump.stdout_seq, 2);

        assert!(pump.apply_frame(exit_frame(pb::SailboxExecStatus::Succeeded, 2)));
        pump.finalize();
        assert_eq!(*shared.high_seq.lock().unwrap(), (2, 0));
        assert!(shared.ended.load(Ordering::SeqCst));
    }

    /// A reader started before any output replays the retained tail in order, then
    /// follows live chunks (including ones that arrive after a reconnect gap), and
    /// stops at Eof once the pump finalizes.
    #[tokio::test]
    async fn exec_reader_follows_replayed_then_live() {
        let shared = test_shared();
        let mut reader = StreamReader {
            shared: shared.clone(),
            which: OutputStream::Stdout,
            cursor: 0,
        };
        let collector = tokio::task::spawn_blocking(move || {
            let mut out = String::new();
            loop {
                match reader.next(Duration::from_millis(50)) {
                    ReadStep::Chunk(piece) => out.push_str(&piece),
                    ReadStep::Eof => return out,
                    ReadStep::Pending => {}
                }
            }
        });

        let mut pump = Pump::new(shared.clone());
        pump.apply_frame(chunk(OutputStream::Stdout, 1, b"hello "));
        tokio::time::sleep(Duration::from_millis(10)).await;
        // A reconnect gap, then the live tail resumes.
        pump.apply_frame(chunk(OutputStream::Stdout, 2, b"world"));
        tokio::time::sleep(Duration::from_millis(10)).await;
        pump.apply_frame(exit_frame(pb::SailboxExecStatus::Succeeded, 2));
        pump.finalize();

        assert_eq!(collector.await.unwrap(), "hello world");
    }

    /// A reader created after output is already buffered still replays the
    /// retained tail from the start (cursor 0), then sees Eof.
    #[test]
    fn exec_reader_started_late_replays_retained_tail() {
        let shared = test_shared();
        let mut pump = Pump::new(shared.clone());
        pump.apply_frame(chunk(OutputStream::Stdout, 1, b"early "));
        pump.apply_frame(chunk(OutputStream::Stdout, 2, b"output"));
        pump.finalize();

        // Construct the reader only now, against an already-populated, ended ring.
        let mut reader = shared_reader(&shared, OutputStream::Stdout);
        let mut out = String::new();
        loop {
            match reader.next(Duration::from_millis(50)) {
                ReadStep::Chunk(piece) => out.push_str(&piece),
                ReadStep::Eof => break,
                ReadStep::Pending => panic!("ended ring should not return Pending"),
            }
        }
        assert_eq!(out, "early output");
    }

    fn shared_reader(shared: &Arc<ExecShared>, which: OutputStream) -> StreamReader {
        StreamReader {
            shared: shared.clone(),
            which,
            cursor: 0,
        }
    }

    use proptest::prelude::*;

    /// Bytes that stress the UTF-8 decoder: random bytes (mostly invalid
    /// sequences) interleaved with the bytes of real Unicode strings (valid
    /// multibyte chars). Split points across these exercise mid-char breaks.
    fn utf8ish_bytes() -> impl Strategy<Value = Vec<u8>> {
        prop_oneof![
            proptest::collection::vec(any::<u8>(), 0..256),
            any::<String>().prop_map(String::into_bytes),
        ]
    }

    proptest! {
        /// Under the cap the ring is lossless: it keeps every byte in order,
        /// reports no drop, and its accounting matches the input exactly.
        #[test]
        fn ring_without_overflow_is_lossless(
            pieces in proptest::collection::vec(any::<String>(), 0..32)
        ) {
            let concat: String = pieces.concat();
            prop_assume!(concat.len() <= STREAM_BUFFER_CAP_BYTES);
            let mut ring = Ring::default();
            for piece in &pieces {
                ring.append(piece.clone());
            }
            prop_assert!(!ring.dropped);
            prop_assert_eq!(ring.size, concat.len());
            prop_assert_eq!(ring.tail(), concat);
        }
    }

    proptest! {
        // Each case allocates ~1 MiB, so keep the case count modest.
        #![proptest_config(ProptestConfig::with_cases(48))]

        /// Once total output exceeds the cap, the ring drops oldest bytes: it
        /// stays within the cap, never splits a multibyte char into a U+FFFD
        /// artifact, and its retained bytes are always a suffix of everything
        /// appended (drops only ever come off the front).
        #[test]
        fn ring_eviction_keeps_valid_suffix_under_cap(
            overflow in 1usize..8192,
            tail_pieces in proptest::collection::vec("[a-z]{0,64}", 0..8),
        ) {
            // A head of 2-byte 'é' that overruns the cap, so the clip path lands
            // inside a multibyte char (cap is even; 2 * count = cap + 2*overflow).
            let head = "é".repeat(STREAM_BUFFER_CAP_BYTES / 2 + overflow);
            let mut full = head.clone();
            let mut ring = Ring::default();
            ring.append(head);
            for piece in &tail_pieces {
                ring.append(piece.clone());
                full.push_str(piece);
            }
            prop_assert!(ring.dropped);
            prop_assert!(ring.size <= STREAM_BUFFER_CAP_BYTES);
            let tail = ring.tail();
            // Clipping advances to a char boundary and slices the valid str, so
            // it never manufactures a replacement char from a split lead byte.
            let has_replacement_char = tail.contains('\u{FFFD}');
            prop_assert!(!has_replacement_char);
            prop_assert!(full.as_bytes().ends_with(tail.as_bytes()));
        }
    }

    proptest! {
        /// The incremental decoder is chunk-boundary invariant: decoding a byte
        /// stream split at arbitrary points yields the same string as decoding
        /// it whole, and that string equals a lossy decode of the full input.
        #[test]
        fn decoder_is_chunk_boundary_invariant(
            data in utf8ish_bytes(),
            cuts in proptest::collection::vec(any::<prop::sample::Index>(), 0..16),
        ) {
            let mut whole = Utf8Decoder::default();
            let mut whole_out = whole.decode(&data);
            whole_out.push_str(&whole.flush());

            let mut bounds: Vec<usize> = cuts.iter().map(|c| c.index(data.len() + 1)).collect();
            bounds.sort_unstable();
            let mut dec = Utf8Decoder::default();
            let mut chunked_out = String::new();
            let mut prev = 0;
            for bound in bounds.into_iter().chain(std::iter::once(data.len())) {
                let bound = bound.clamp(prev, data.len());
                chunked_out.push_str(&dec.decode(&data[prev..bound]));
                prev = bound;
            }
            chunked_out.push_str(&dec.flush());

            prop_assert_eq!(&chunked_out, &whole_out);
            prop_assert_eq!(whole_out, String::from_utf8_lossy(&data).into_owned());
        }
    }

    proptest! {
        /// The per-stream high-water seq is the running max of the seqs seen on
        /// that stream and only ever advances, regardless of frame order, so a
        /// replayed or out-of-order tail after a reconnect can't lower the
        /// resume point, and the two streams are tracked independently.
        #[test]
        fn pump_seq_is_monotonic_running_max_per_stream(
            frames in proptest::collection::vec(
                (any::<bool>(), 0i64..10_000, proptest::collection::vec(any::<u8>(), 0..8)),
                0..64,
            )
        ) {
            let mut pump = Pump::new(test_shared());
            let (mut expect_out, mut expect_err) = (0i64, 0i64);
            let (mut prev_out, mut prev_err) = (0i64, 0i64);
            for (is_stderr, seq, data) in frames {
                let which = if is_stderr { OutputStream::Stderr } else { OutputStream::Stdout };
                pump.apply_frame(chunk(which, seq, &data));
                if is_stderr {
                    expect_err = expect_err.max(seq);
                } else {
                    expect_out = expect_out.max(seq);
                }
                prop_assert_eq!(pump.stdout_seq, expect_out);
                prop_assert_eq!(pump.stderr_seq, expect_err);
                prop_assert!(pump.stdout_seq >= prev_out);
                prop_assert!(pump.stderr_seq >= prev_err);
                prev_out = pump.stdout_seq;
                prev_err = pump.stderr_seq;
            }
        }
    }
}