sail-rs 0.3.1

//! Streaming exec engine: a running command in a sailbox with live output.
//!
//! [`ExecProcess::start`] opens a server-streaming exec request. The first frame
//! reports that the command has started (carrying the durable
//! `exec_request_id`); chunk frames carry live stdout/stderr; a terminal 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 fetching the
//! authoritative buffered result. Stdin rides a separate offset-idempotent unary
//! request, 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`].
#[doc(hidden)]
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 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, exactly as the command wrote it: the
    /// live stream is a byte pipe (escape sequences and binary payloads
    /// included), not decoded text. String-typed conveniences decode at the
    /// edge ([`ExecResult`], the bindings' str iterators).
    Chunk(Vec<u8>),
    /// 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, lossily decoded as UTF-8 (the live byte stream is
    /// unmodified; the decode happens only here). For a pty exec that was
    /// reattached, this is the last screen repaint plus the output after it,
    /// not a full transcript.
    pub stdout: String,
    /// Buffered stderr, lossily decoded as UTF-8 (see `stdout`).
    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,
    /// Total bytes the command wrote to stdout over its whole run, including
    /// bytes truncation dropped from the buffered `stdout` field above. `0` when
    /// unknown (no exit was observed on the stream, or an older guest). Subtract
    /// what a consumer actually saw to learn how much was lost.
    pub stdout_total_bytes: i64,
    /// Total bytes the command wrote to stderr over its whole run (see
    /// `stdout_total_bytes`).
    pub stderr_total_bytes: i64,
}

/// 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.
    #[doc(hidden)]
    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.
    #[doc(hidden)]
    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,
    /// Extra environment for the command, applied for pty and non-pty execs
    /// alike. Entries override the guest's defaults (including `LANG`) and the
    /// image env. A few reserved variables that identify the sailbox (such as
    /// `SAILBOX_ID`) cannot be overridden. For pty execs the terminal variables
    /// (`COLORTERM`, `LANG`, `LC_*`, `TERM_PROGRAM`) are auto-forwarded from the
    /// local environment for keys not set here.
    pub env: Vec<(String, String)>,
    /// 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,
            env: Vec::new(),
            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('\'', "'\\''"))
}

/// Local env vars auto-forwarded to pty execs so terminal programs render
/// correctly (truecolor detection, locale-driven width math). TERM rides the
/// dedicated `term` field, not this list.
const PTY_ENV_WHITELIST: [&str; 3] = ["COLORTERM", "LANG", "TERM_PROGRAM"];

fn pty_env_whitelisted(key: &str) -> bool {
    PTY_ENV_WHITELIST.contains(&key) || key.starts_with("LC_")
}

/// Snapshot the local environment filtered to the pty forwarding whitelist.
pub(crate) fn pty_forward_env() -> Vec<(String, String)> {
    // vars_os, not vars: std::env::vars panics on any non-Unicode entry in the
    // inherited environment, even one unrelated to the whitelist. Entries that
    // do not decode cannot ride a proto string map anyway, so they are skipped.
    pty_forward_env_from(
        std::env::vars_os()
            .filter_map(|(key, value)| Some((key.into_string().ok()?, value.into_string().ok()?))),
    )
}

fn pty_forward_env_from(vars: impl Iterator<Item = (String, String)>) -> Vec<(String, String)> {
    vars.filter(|(key, _)| pty_env_whitelisted(key)).collect()
}

/// Validate user-supplied env pairs into the wire map. Values are free-form;
/// keys must be non-empty and free of `=` and NUL (execve constraints). Every
/// binding's user env funnels through here (the Rust client and the bindings
/// that build `ExecParams` directly), so a key like `"A=B"` fails loudly
/// instead of silently becoming a different variable in the guest.
#[doc(hidden)]
pub fn encode_env(
    pairs: &[(String, String)],
) -> Result<std::collections::HashMap<String, String>, SailError> {
    let mut env = std::collections::HashMap::with_capacity(pairs.len());
    for (key, value) in pairs {
        // The name must be a portable identifier and the value must carry no NUL
        // (execve cannot represent either). is_portable_env_name already rejects
        // '=', whitespace, and NUL in the name, so only the value needs a guard.
        if !is_portable_env_name(key) || value.contains('\0') {
            return Err(SailError::InvalidArgument {
                message: format!("invalid env entry {key:?}"),
            });
        }
        env.insert(key.clone(), value.clone());
    }
    Ok(env)
}

/// Whether `name` is a portable environment variable name: a non-empty run of
/// `[A-Za-z_][A-Za-z0-9_]*`. Rejects a leading digit, whitespace, `=`, a NUL, or
/// any other character the guest could not represent (or a shell could not read
/// back) as an environment entry.
fn is_portable_env_name(name: &str) -> bool {
    let mut chars = name.chars();
    match chars.next() {
        Some(c) if c.is_ascii_alphabetic() || c == '_' => {}
        _ => return false,
    }
    chars.all(|c| c.is_ascii_alphanumeric() || c == '_')
}

/// 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.
#[doc(hidden)]
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,
    /// Extra environment for the command as wire-ready KEY=VALUE entries,
    /// resolved once at launch (including the pty terminal whitelist) and
    /// resent verbatim on every reconnect.
    pub env: std::collections::HashMap<String, String>,
    /// 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)>,
}

/// Drop-oldest byte ring mirroring the server output ring. Appends never
/// block; past the cap the oldest bytes are dropped (byte-exact) 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<Vec<u8>>,
    first_idx: usize,
    size: usize,
    dropped: bool,
    /// Monotonic count of in-place front-piece clips. A clip drops bytes from
    /// the piece at `first_idx` without advancing it, so a reader parked on
    /// that piece cannot see the loss through `first_idx` alone; it compares
    /// this instead.
    front_clips: u64,
    /// Monotonic count of `reset_to` repaints. A repaint advances `first_idx`
    /// past a reader's cursor like an eviction, but it supersedes those bytes
    /// with a fresh screen instead of losing them, so a reader compares this to
    /// tell a heal from a fall-behind drop.
    resets: u64,
}

impl Ring {
    fn append(&mut self, data: Vec<u8>) {
        self.size += data.len();
        self.pieces.push(data);
        while self.size > STREAM_BUFFER_CAP_BYTES {
            let overflow = self.size - STREAM_BUFFER_CAP_BYTES;
            if self.pieces[0].len() <= overflow {
                self.size -= self.pieces[0].len();
                self.pieces.remove(0);
                self.first_idx += 1;
            } else {
                self.pieces[0].drain(..overflow);
                self.size -= overflow;
                self.front_clips += 1;
            }
            self.dropped = true;
        }
    }

    /// Replace the retained content with a pty screen repaint. Advancing
    /// `first_idx` past the old pieces makes every attached reader (cursor
    /// below it) skip straight to the repaint, and a late reader replays only
    /// the repaint. `dropped` is cleared: the repaint supersedes everything
    /// the ring ever dropped, so a healed session must not read as truncated
    /// (which would force `wait()` into the server fallback). `append`
    /// re-latches it only if the repaint itself overflows.
    fn reset_to(&mut self, repaint: Vec<u8>) {
        self.first_idx += self.pieces.len();
        self.pieces.clear();
        self.size = 0;
        self.dropped = false;
        self.resets += 1;
        if !repaint.is_empty() {
            self.append(repaint);
        }
    }

    fn tail(&self) -> Vec<u8> {
        self.pieces.concat()
    }
}

/// Lossily decode a ring's retained bytes for the string-typed [`ExecResult`].
/// The only place live output becomes text in the core. NUL is replaced too:
/// it is valid UTF-8 that `from_utf8_lossy` keeps, but the text result is the
/// client twin of the guest's persisted tail (which replaces NUL with U+FFFD
/// for its Postgres text column), so the two agree. The raw byte readers keep NUL.
fn lossy_tail(ring: &Ring) -> String {
    String::from_utf8_lossy(&ring.tail()).replace('\0', "\u{FFFD}")
}

#[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,
    stdout_total_bytes: i64,
    stderr_total_bytes: 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()
        };
        // Resolve the pty terminal-env whitelist once at launch (caller-supplied
        // keys win); params is reused verbatim on every reconnect.
        if params.pty {
            for (key, value) in pty_forward_env() {
                params.env.entry(key).or_insert(value);
            }
        }
        let (exec_request_id, stream) = submit(
            &worker, &params, /* stdout_resume_seq */ 0, /* stderr_resume_seq */ 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,
            dropped: false,
            reset: false,
            seen_front_clips: 0,
            seen_resets: 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,
        }
    }

    /// The locally buffered raw bytes of one stream (the byte-typed twin of the
    /// lossily decoded [`ExecResult`] fields): exactly what the readers have
    /// been fed, capped drop-oldest. A byte-count reconciliation against what a
    /// consumer already printed must use this, not the decoded strings, whose
    /// lengths diverge from the raw stream on invalid UTF-8.
    #[doc(hidden)]
    pub fn buffered_output(&self, which: OutputStream) -> Vec<u8> {
        lock(&self.shared.state).ring(which).tail()
    }

    /// 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.
    #[doc(hidden)]
    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 server fallback only recovers a MISSING ENDING: a stream with no
        // Exit, or one that fell short of the exit's high-water seq (a migration
        // replay that did not deliver the tail). It does NOT help a stream that
        // is complete but merely truncated at the front — the server's persisted
        // tail is smaller than the local ring, so falling back there returns a
        // worse result and burns a WaitSailboxExec RPC. Truncation is reported
        // honestly on the local result instead.
        let incomplete = exit
            .as_ref()
            .is_some_and(|exit| exit.stdout_seq > high_out || exit.stderr_seq > high_err);
        // Whether the buffered `stdout`/`stderr` fields dropped their oldest
        // bytes — the guest ring overflowed, or the local ring evicted output
        // the reader may already have consumed. Complete streams still report
        // it so a caller knows the convenience field is a tail, not the whole.
        let stdout_truncated_flag = exit
            .as_ref()
            .is_some_and(|exit| exit.stdout_truncated || out_dropped);
        let stderr_truncated_flag = exit
            .as_ref()
            .is_some_and(|exit| exit.stderr_truncated || err_dropped);

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

        // Total bytes the command produced on each stream (0 when no exit was
        // observed on the stream, or the guest is too old to report it). A caller
        // subtracts what it saw to learn how much truncation dropped.
        let stdout_total_bytes = exit.as_ref().map_or(0, |exit| exit.stdout_total_bytes);
        let stderr_total_bytes = exit.as_ref().map_or(0, |exit| exit.stderr_total_bytes);

        if exit.is_none() || 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,
                        // WaitSailboxExec carries no byte totals; 0 = unknown.
                        stdout_total_bytes: 0,
                        stderr_total_bytes: 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 = lossy_tail(&state.stderr);
                    if witnessed.status == pb::SailboxExecStatus::Failed as i32
                        && !witnessed.error_message.is_empty()
                    {
                        stderr = witnessed.error_message.clone();
                    }
                    return Ok(ExecResult {
                        stdout: lossy_tail(&state.stdout),
                        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,
                        stdout_total_bytes,
                        stderr_total_bytes,
                    });
                }
            }
            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,
                stdout_total_bytes,
                stderr_total_bytes,
            });
        }

        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 = lossy_tail(&state.stderr);
        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: lossy_tail(&state.stdout),
            stderr,
            exit_code: exit.exit_code,
            timed_out: exit.timed_out,
            stdout_truncated: stdout_truncated_flag,
            stderr_truncated: stderr_truncated_flag,
            stdout_complete,
            stderr_complete,
            stdout_total_bytes,
            stderr_total_bytes,
        })
    }

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

    /// Ask a `pty` exec to re-emit its current screen as a Snapshot on the live
    /// stream. A client whose local buffer dropped output (it fell behind a
    /// fast producer) calls this to repaint instead of rendering a torn tail;
    /// the command keeps running detached. Advisory and best-effort like
    /// [`resize`](Self::resize): an unknown, finished, or non-pty exec is a
    /// server no-op, and a transient error is swallowed (the client re-requests
    /// if it is still behind).
    pub async fn resync(&self) {
        let message = pb::ResyncSailboxExecRequest {
            sailbox_id: self.shared.params.sailbox_id.clone(),
            exec_request_id: self.exec_request_id.clone(),
        };
        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.resync_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, /* eof */ 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, &[], /* eof */ 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,
    dropped: bool,
    /// Set when the ring was reset to a repaint since the last read. Surfaced
    /// via took_reset so an interactive consumer drops its stale local backlog
    /// before rendering the repaint, rather than leaving it stuck behind bytes
    /// the terminal will never finish draining.
    reset: bool,
    /// The ring's `front_clips` value this reader has already accounted for, so
    /// an in-place clip of the piece it is parked on registers as a drop once.
    seen_front_clips: u64,
    /// The ring's `reset_to` count this reader has accounted for. A repaint
    /// advances `first_idx` past the cursor like an eviction, but reading the
    /// repaint heals the screen, so it must not register as a fall-behind drop.
    seen_resets: u64,
}

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);
            // Reconcile with the ring head. A reset_to repaint replaced the ring
            // and supersedes whatever was skipped, so it is a heal surfaced via
            // took_reset, not a drop; it also supersedes a drop an earlier read
            // had already latched. Adopt the ring's own dropped flag rather than
            // clearing unconditionally: reset_to clears it, so it is false for a
            // clean heal, but re-latches if the repaint itself was then evicted
            // by later output before this read. In that case the reader is about
            // to hand back a torn post-repaint suffix, so it must still report a
            // drop for the consumer to resync. Otherwise a cursor below the head
            // means the ring evicted chunks this reader had not consumed, and a
            // front-clip trims the piece at first_idx in place without advancing
            // it; both latch a drop so an interactive consumer can repaint
            // (took_drop).
            if ring.resets != self.seen_resets {
                self.reset = true;
                self.dropped = ring.dropped;
                if self.cursor < ring.first_idx {
                    self.cursor = ring.first_idx;
                }
            } else if self.cursor < ring.first_idx {
                self.cursor = ring.first_idx;
                self.dropped = true;
            } else if self.cursor == ring.first_idx && ring.front_clips != self.seen_front_clips {
                self.dropped = true;
            }
            self.seen_front_clips = ring.front_clips;
            self.seen_resets = ring.resets;
            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;
            }
        }
    }

    /// The next retained chunk if one is already buffered, without blocking.
    /// `None` means the ring is momentarily drained (not that the stream
    /// ended); callers batch-draining an interactive stream use it to flush
    /// several chunks in one write.
    pub fn try_next(&mut self) -> Option<Vec<u8>> {
        let state = lock(&self.shared.state);
        let ring = state.ring(self.which);
        if ring.resets != self.seen_resets {
            self.reset = true;
            // Adopt the ring's dropped flag: a clean repaint clears it, but a
            // repaint evicted by later output before this read leaves it set, so
            // the torn suffix still reports a drop. See next() for the full note.
            self.dropped = ring.dropped;
            if self.cursor < ring.first_idx {
                self.cursor = ring.first_idx;
            }
        } else if self.cursor < ring.first_idx {
            self.cursor = ring.first_idx;
            self.dropped = true;
        } else if self.cursor == ring.first_idx && ring.front_clips != self.seen_front_clips {
            self.dropped = true;
        }
        self.seen_front_clips = ring.front_clips;
        self.seen_resets = ring.resets;
        if self.cursor < ring.first_idx + ring.pieces.len() {
            let piece = ring.pieces[self.cursor - ring.first_idx].clone();
            self.cursor += 1;
            Some(piece)
        } else {
            None
        }
    }

    /// Whether the ring evicted output this reader had not yet consumed since
    /// the last call, clearing the flag. An interactive consumer uses it to
    /// trigger a screen repaint ([`ExecProcess::resync`]) after falling behind.
    pub fn took_drop(&mut self) -> bool {
        std::mem::take(&mut self.dropped)
    }

    /// Whether the ring was reset to a repaint (a Snapshot superseded the
    /// stream) since the last call, clearing the flag. An interactive consumer
    /// drops any stale terminal-local backlog on this so the repaint it is about
    /// to read renders at once instead of stuck behind bytes the terminal will
    /// never finish draining.
    pub fn took_reset(&mut self) -> bool {
        std::mem::take(&mut self.reset)
    }
}

/// 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 of raw bytes, 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<Vec<u8>> {
        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(),
            env: params.env.clone(),
        };
        let request =
            worker.request_for(message, &params.extra_metadata, /* timeout */ 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). Split out from the transport
/// loop so the resume/replay 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_seq: i64,
    stderr_seq: i64,
}

impl Pump {
    fn new(shared: Arc<ExecShared>) -> Pump {
        Pump {
            shared,
            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 — a server that
    /// replays an already-seen tail after reconnect can't lower the resume point
    /// — with one deliberate exception: a pty Snapshot assigns its basis, since
    /// the repaint supersedes everything before it.
    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 which = if is_stderr {
                    self.stderr_seq = self.stderr_seq.max(chunk.seq);
                    OutputStream::Stderr
                } else {
                    self.stdout_seq = self.stdout_seq.max(chunk.seq);
                    OutputStream::Stdout
                };
                if !chunk.data.is_empty() {
                    let mut state = lock(&self.shared.state);
                    match which {
                        OutputStream::Stdout => state.stdout.append(chunk.data),
                        OutputStream::Stderr => state.stderr.append(chunk.data),
                    }
                    self.shared.cond.notify_all();
                    self.shared.data_notify.notify_waiters();
                }
                false
            }
            Some(pb::stream_sailbox_exec_response::Frame::Snapshot(snap)) => {
                // A pty screen resync (reattach or server-side overflow
                // recovery): the repaint replaces the retained stream and seq
                // accounting continues from the basis, so Exit-completeness
                // math stays honest without any snapshot-specific logic in
                // wait() or the readers.
                self.stdout_seq = snap.stdout_seq_basis;
                let mut state = lock(&self.shared.state);
                state.stdout.reset_to(snap.repaint);
                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,
                    stdout_total_bytes: exit.stdout_total_bytes,
                    stderr_total_bytes: exit.stderr_total_bytes,
                });
                true
            }
            _ => false,
        }
    }

    /// Publish the high-water seqs, close the rings, and wake every reader and
    /// waiter. Consumes the pump: nothing follows finalize.
    fn finalize(self) {
        {
            let mut state = lock(&self.shared.state);
            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_exec_ring(data: &[u8]) {
    // Append the input split at data-derived boundaries (cut after every odd
    // byte) and as one piece: the retained tail must be identical — the ring is
    // chunk-boundary invariant on raw bytes.
    let mut chunked = Ring::default();
    let mut start = 0;
    for (i, byte) in data.iter().enumerate() {
        if byte & 1 == 1 {
            chunked.append(data[start..=i].to_vec());
            start = i + 1;
        }
    }
    if start < data.len() {
        chunked.append(data[start..].to_vec());
    }
    let mut whole = Ring::default();
    if !data.is_empty() {
        whole.append(data.to_vec());
    }
    assert_eq!(
        chunked.tail(),
        whole.tail(),
        "ring is not chunk-boundary invariant"
    );

    // 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 (byte-exact)
    // and its retained bytes must remain a suffix of everything appended.
    let mut ring = Ring::default();
    ring.append(data.to_vec());
    let filler = vec![b'a'; STREAM_BUFFER_CAP_BYTES + 1 - data.len().min(STREAM_BUFFER_CAP_BYTES)];
    ring.append(filler.clone());
    assert!(
        ring.size <= STREAM_BUFFER_CAP_BYTES,
        "ring exceeded its cap"
    );
    let mut full = data.to_vec();
    full.extend_from_slice(&filler);
    assert!(
        full.ends_with(&ring.tail()),
        "ring tail is not a suffix of the appended bytes"
    );

    // A reset supersedes everything retained: prior pieces become unreachable
    // (first_idx advanced past them) and the tail is exactly the repaint,
    // clipped to the cap.
    let before_reset_pieces = ring.first_idx + ring.pieces.len();
    ring.reset_to(data.to_vec());
    assert!(
        ring.first_idx >= before_reset_pieces,
        "reset left old pieces reachable"
    );
    let expected = &data[data.len().saturating_sub(STREAM_BUFFER_CAP_BYTES)..];
    assert_eq!(
        ring.tail(),
        expected,
        "reset tail is not the capped repaint"
    );
}

#[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_exec_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_exec_ring(sample);
        }
    }

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

    #[test]
    fn ring_clip_is_byte_exact() {
        // Overflow of 1 lands inside the leading 2-byte 'é'. The byte ring
        // clips at the exact byte — split multibyte sequences are the reader's
        // concern (decode at the edge), never the ring's.
        let mut ring = Ring::default();
        let mut data = "é".as_bytes().to_vec();
        data.extend(std::iter::repeat_n(b'a', STREAM_BUFFER_CAP_BYTES - 1));
        ring.append(data);
        assert_eq!(ring.size, STREAM_BUFFER_CAP_BYTES);
        assert!(ring.dropped);
        let tail = ring.tail();
        assert_eq!(tail.len(), STREAM_BUFFER_CAP_BYTES);
        // The clip cut the first byte of 'é'; its continuation byte survives.
        assert_eq!(tail[0], "é".as_bytes()[1]);
    }

    #[test]
    fn ring_reset_to_supersedes_retained_pieces() {
        let mut ring = Ring::default();
        ring.append(b"old output".to_vec());
        ring.append(b"more".to_vec());
        let reachable_end = ring.first_idx + ring.pieces.len();
        ring.reset_to(b"\x1b[2J\x1b[Hrepaint".to_vec());
        assert!(ring.first_idx >= reachable_end);
        assert_eq!(ring.tail(), b"\x1b[2J\x1b[Hrepaint");
        // A catch-up, not a loss: dropped must not latch.
        assert!(!ring.dropped);

        let mut empty = Ring::default();
        empty.append(b"x".to_vec());
        empty.reset_to(Vec::new());
        assert!(empty.tail().is_empty());
    }

    /// A drop latched before the snapshot must not survive it: the repaint
    /// supersedes the lost bytes, and a stale `dropped` would force `wait()`
    /// into the server fallback for a fully healed pty session.
    #[test]
    fn pty_forward_env_filters_to_the_whitelist() {
        let vars = vec![
            ("COLORTERM".to_string(), "truecolor".to_string()),
            ("LC_ALL".to_string(), "en_US.UTF-8".to_string()),
            ("LANG".to_string(), "en_US.UTF-8".to_string()),
            ("TERM_PROGRAM".to_string(), "TestTerm".to_string()),
            ("PATH".to_string(), "/bin".to_string()),
            ("TERM".to_string(), "xterm".to_string()), // rides the term field, not env
            ("SECRET_TOKEN".to_string(), "x".to_string()),
        ];
        let forwarded = pty_forward_env_from(vars.into_iter());
        let keys: Vec<&str> = forwarded.iter().map(|(k, _)| k.as_str()).collect();
        assert_eq!(keys, ["COLORTERM", "LC_ALL", "LANG", "TERM_PROGRAM"]);
    }

    #[test]
    fn encode_env_rejects_malformed_keys() {
        for (key, value) in [
            ("", "v"),
            ("A=B", "v"),
            ("NUL\0KEY", "v"),
            ("K", "nul\0value"),
            // Non-portable names: leading digit, whitespace, punctuation.
            ("1FOO", "v"),
            ("FO O", "v"),
            ("FOO-BAR", "v"),
            ("FOO.BAR", "v"),
        ] {
            let pairs = vec![(key.to_string(), value.to_string())];
            assert!(
                encode_env(&pairs).is_err(),
                "expected rejection for {key:?}={value:?}"
            );
        }
        // A leading-underscore name and an opaque value (including '=') are fine.
        let ok = encode_env(&[
            ("_FOO".to_string(), "bar=baz".to_string()),
            ("LC_ALL".to_string(), "C.UTF-8".to_string()),
        ])
        .unwrap();
        assert_eq!(ok.get("_FOO").map(String::as_str), Some("bar=baz"));
    }

    #[test]
    fn ring_reset_to_clears_prior_dropped() {
        let mut ring = Ring::default();
        ring.append(vec![b'a'; STREAM_BUFFER_CAP_BYTES + 1]);
        assert!(ring.dropped);
        ring.reset_to(b"repaint".to_vec());
        assert!(!ring.dropped);
        assert_eq!(ring.tail(), b"repaint");

        // An over-cap repaint re-latches through append's normal path.
        let mut over = Ring::default();
        over.append(vec![b'b'; STREAM_BUFFER_CAP_BYTES + 1]);
        over.reset_to(vec![b'c'; STREAM_BUFFER_CAP_BYTES + 1]);
        assert!(over.dropped);
        assert_eq!(over.size, STREAM_BUFFER_CAP_BYTES);
    }

    #[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,
                env: std::collections::HashMap::default(),
                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,
                    ..Default::default()
                },
            )),
        }
    }

    /// The resume state machine: bytes split across a mid-stream break arrive
    /// verbatim (a mid-char break is just two chunks; the edge decode heals it),
    /// 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_bytes_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(), b"ab\xc3");
        // 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; the ring concatenates the raw bytes, and the
        // string edge lossy-decodes them whole.
        assert!(!pump.apply_frame(chunk(OutputStream::Stdout, 2, b"\xa9cd")));
        assert_eq!(
            shared.state.lock().unwrap().stdout.tail(),
            "abécd".as_bytes()
        );
        assert_eq!(lossy_tail(&shared.state.lock().unwrap().stdout), "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));
    }

    fn snapshot_frame(repaint: &[u8], basis: i64) -> pb::StreamSailboxExecResponse {
        pb::StreamSailboxExecResponse {
            frame: Some(pb::stream_sailbox_exec_response::Frame::Snapshot(
                pb::SailboxExecSnapshot {
                    repaint: repaint.to_vec(),
                    stdout_seq_basis: basis,
                },
            )),
        }
    }

    /// A pty Snapshot supersedes the retained stream: readers skip to the
    /// repaint, a late reader sees only the repaint, the seq high-water is
    /// assigned to the basis, and Exit-completeness math continues from it.
    #[tokio::test]
    async fn snapshot_resets_ring_seq_basis_and_skips_readers() {
        let shared = test_shared();
        let mut pump = Pump::new(shared.clone());
        assert!(!pump.apply_frame(chunk(OutputStream::Stdout, 1, b"pre-disconnect ")));
        assert!(!pump.apply_frame(chunk(OutputStream::Stdout, 2, b"tail")));

        // The reattach answers with a repaint based at the guest's high seq.
        assert!(!pump.apply_frame(snapshot_frame(b"\x1b[2J\x1b[Hscreen", 7)));
        assert_eq!(pump.stdout_seq, 7);

        // A late reader replays only the repaint, then the post-snapshot chunk.
        assert!(!pump.apply_frame(chunk(OutputStream::Stdout, 8, b" after")));
        assert!(pump.apply_frame(exit_frame(pb::SailboxExecStatus::Succeeded, 8)));
        pump.finalize();

        let mut reader = shared_reader(&shared, OutputStream::Stdout);
        let mut out = Vec::new();
        loop {
            match reader.next(Duration::from_millis(50)) {
                ReadStep::Chunk(piece) => out.extend_from_slice(&piece),
                ReadStep::Eof => break,
                ReadStep::Pending => panic!("ended ring should not return Pending"),
            }
        }
        assert_eq!(out, b"\x1b[2J\x1b[Hscreen after");
        // Completeness: exit.stdout_seq (8) <= published high seq (8).
        assert_eq!(*shared.high_seq.lock().unwrap(), (8, 0));
    }

    #[test]
    fn snapshot_with_empty_repaint_is_reset_only() {
        let shared = test_shared();
        let mut pump = Pump::new(shared.clone());
        pump.apply_frame(chunk(OutputStream::Stdout, 3, b"stale"));
        pump.apply_frame(snapshot_frame(b"", 3));
        assert_eq!(pump.stdout_seq, 3);
        let state = lock(&shared.state);
        assert!(state.stdout.tail().is_empty());
        assert!(!state.stdout.dropped);
    }

    /// 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,
            dropped: false,
            reset: false,
            seen_front_clips: 0,
            seen_resets: 0,
        };
        let collector = tokio::task::spawn_blocking(move || {
            let mut out = Vec::new();
            loop {
                match reader.next(Duration::from_millis(50)) {
                    ReadStep::Chunk(piece) => out.extend_from_slice(&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(), b"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 = Vec::new();
        loop {
            match reader.next(Duration::from_millis(50)) {
                ReadStep::Chunk(piece) => out.extend_from_slice(&piece),
                ReadStep::Eof => break,
                ReadStep::Pending => panic!("ended ring should not return Pending"),
            }
        }
        assert_eq!(out, b"early output");
    }

    // A reader that falls behind a ring overflow skips the evicted chunks and
    // reports the drop once (then clears it), and try_next batch-drains without
    // blocking. This is what the interactive bridge keys its repaint request on.
    #[test]
    fn reader_reports_drop_and_batch_drains() {
        let shared = test_shared();
        {
            // A small head chunk plus a cap-sized chunk overflows the ring,
            // fully evicting the head (advancing first_idx past it).
            let mut state = lock(&shared.state);
            state.stdout.append(b"HEAD".to_vec());
            state.stdout.append(vec![b'x'; STREAM_BUFFER_CAP_BYTES]);
            state.ended = true;
        }
        let mut reader = shared_reader(&shared, OutputStream::Stdout);

        // The reader skips the evicted head and reports the drop once.
        let first = reader.next(Duration::from_millis(50));
        let ReadStep::Chunk(first) = first else {
            panic!("expected the retained chunk, got {first:?}");
        };
        assert!(
            reader.took_drop(),
            "the overflow evicted an unconsumed chunk"
        );
        assert!(!reader.took_drop(), "took_drop clears the latch");

        // The retained content is the surviving chunk; the head is gone.
        assert_eq!(first, vec![b'x'; STREAM_BUFFER_CAP_BYTES]);
        assert!(
            reader.try_next().is_none(),
            "a drained ring yields None without blocking"
        );
    }

    #[test]
    fn reader_does_not_report_a_reset_repaint_as_a_drop() {
        let shared = test_shared();
        let mut reader = shared_reader(&shared, OutputStream::Stdout);
        {
            let mut state = lock(&shared.state);
            state.stdout.append(b"one".to_vec());
        }
        // Consume the first chunk cleanly: no drop yet.
        assert!(matches!(
            reader.next(Duration::from_millis(50)),
            ReadStep::Chunk(_)
        ));
        assert!(!reader.took_drop(), "a clean read latches no drop");
        assert!(!reader.took_reset(), "a clean read is not a reset");

        {
            // More live output the reader has not read, then a repaint that
            // supersedes the ring and advances first_idx past the cursor. The
            // reader must read the repaint as a heal, not report a new drop:
            // otherwise the pump discards the repaint and loops on resyncs.
            let mut state = lock(&shared.state);
            state.stdout.append(b"two".to_vec());
            state.stdout.append(b"three".to_vec());
            state.stdout.reset_to(b"REPAINT".to_vec());
            state.ended = true;
        }
        let repaint = reader.next(Duration::from_millis(50));
        let ReadStep::Chunk(repaint) = repaint else {
            panic!("expected the repaint chunk, got {repaint:?}");
        };
        assert_eq!(repaint, b"REPAINT");
        assert!(
            !reader.took_drop(),
            "reading a reset repaint is a heal, not a fall-behind drop"
        );
        assert!(
            reader.took_reset(),
            "the reset repaint is surfaced as a took_reset event so the pump can \
             drop stale terminal-local backlog buffered ahead of it"
        );
        assert!(!reader.took_reset(), "took_reset clears the latch");
    }

    #[test]
    fn reader_reset_supersedes_a_previously_latched_drop() {
        let shared = test_shared();
        {
            // A small head plus a cap-sized chunk overflows the ring, evicting
            // the head so the first read below latches a drop.
            let mut state = lock(&shared.state);
            state.stdout.append(b"HEAD".to_vec());
            state.stdout.append(vec![b'x'; STREAM_BUFFER_CAP_BYTES]);
        }
        let mut reader = shared_reader(&shared, OutputStream::Stdout);

        // Read the surviving chunk. The drop is latched but not yet observed,
        // the way the output pump reads a chunk and only checks took_drop at the
        // end of its loop iteration.
        assert!(matches!(
            reader.next(Duration::from_millis(50)),
            ReadStep::Chunk(_)
        ));

        {
            // A repaint supersedes the stream before that pending drop is
            // observed. The repaint heals the drop too, so the drop latch must
            // not survive to trigger a resync that would discard the repaint.
            let mut state = lock(&shared.state);
            state.stdout.reset_to(b"REPAINT".to_vec());
            state.ended = true;
        }
        let repaint = reader.next(Duration::from_millis(50));
        let ReadStep::Chunk(repaint) = repaint else {
            panic!("expected the repaint chunk, got {repaint:?}");
        };
        assert_eq!(repaint, b"REPAINT");
        assert!(reader.took_reset(), "the repaint is surfaced as a reset");
        assert!(
            !reader.took_drop(),
            "the reset superseded the stale drop latch, so no redundant resync \
             clobbers the repaint"
        );
    }

    #[test]
    fn reader_reports_a_drop_when_output_evicts_the_repaint_before_it_is_read() {
        let shared = test_shared();
        let mut reader = shared_reader(&shared, OutputStream::Stdout);
        {
            let mut state = lock(&shared.state);
            state.stdout.append(b"one".to_vec());
        }
        // Consume the first chunk cleanly: no drop, no reset yet.
        assert!(matches!(
            reader.next(Duration::from_millis(50)),
            ReadStep::Chunk(_)
        ));
        assert!(!reader.took_drop());
        assert!(!reader.took_reset());

        {
            // A repaint resets the ring, then a burst larger than the cap evicts
            // that repaint before the reader observes the reset. The reader is
            // now about to hand the terminal a torn post-repaint suffix, not the
            // repaint. It must still report a drop so the pump resyncs for a
            // fresh repaint, rather than leaving the terminal on an arbitrary
            // fragment when the stream then goes idle.
            let mut state = lock(&shared.state);
            state.stdout.reset_to(b"REPAINT".to_vec());
            state.stdout.append(vec![b'x'; STREAM_BUFFER_CAP_BYTES + 1]);
            state.ended = true;
        }
        let chunk = reader.next(Duration::from_millis(50));
        let ReadStep::Chunk(chunk) = chunk else {
            panic!("expected the retained suffix, got {chunk:?}");
        };
        assert_ne!(chunk, b"REPAINT", "the repaint was evicted by the burst");
        assert!(reader.took_reset(), "a reset did occur");
        assert!(
            reader.took_drop(),
            "the repaint was evicted after the reset, so the reader reports a \
             drop and the pump resyncs instead of showing a torn suffix"
        );
    }

    #[test]
    fn reader_try_next_batch_drains_buffered_chunks() {
        let shared = test_shared();
        {
            let mut state = lock(&shared.state);
            state.stdout.append(b"one".to_vec());
            state.stdout.append(b"two".to_vec());
            state.stdout.append(b"three".to_vec());
            state.ended = true;
        }
        let mut reader = shared_reader(&shared, OutputStream::Stdout);

        // next() takes the first chunk; try_next() drains the rest in one batch
        // without blocking, then reports the ring is momentarily empty.
        let first = reader.next(Duration::from_millis(50));
        let ReadStep::Chunk(first) = first else {
            panic!("expected the first buffered chunk, got {first:?}");
        };
        assert_eq!(first, b"one");
        assert_eq!(reader.try_next(), Some(b"two".to_vec()));
        assert_eq!(reader.try_next(), Some(b"three".to_vec()));
        assert!(reader.try_next().is_none(), "the ring is drained");
        assert!(!reader.took_drop(), "no eviction, so no drop is latched");
    }

    #[test]
    fn reader_reports_a_front_clip_as_a_drop() {
        let shared = test_shared();
        {
            // Fill the ring to exactly the cap with one piece the reader parks on.
            let mut state = lock(&shared.state);
            state.stdout.append(vec![b'a'; STREAM_BUFFER_CAP_BYTES]);
        }
        let mut reader = shared_reader(&shared, OutputStream::Stdout);
        {
            // A small append overflows by less than the front piece, so the ring
            // trims that piece in place rather than evicting it: first_idx holds.
            let mut state = lock(&shared.state);
            state.stdout.append(b"tail".to_vec());
            state.ended = true;
            assert_eq!(
                state.stdout.first_idx, 0,
                "the front piece was clipped, not evicted"
            );
        }
        // The reader is still parked on the clipped piece, so it must latch the
        // drop even though first_idx never moved.
        let step = reader.next(Duration::from_millis(50));
        let ReadStep::Chunk(_) = step else {
            panic!("expected the clipped front piece, got {step:?}");
        };
        assert!(
            reader.took_drop(),
            "an in-place front clip of the parked piece is a drop"
        );
    }

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

    use proptest::prelude::*;

    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(proptest::collection::vec(any::<u8>(), 0..512), 0..32)
        ) {
            let concat: Vec<u8> = 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 byte-exactly at the cap and its retained bytes are always a
        /// suffix of everything appended (drops only ever come off the front).
        #[test]
        fn ring_eviction_keeps_byte_suffix_at_cap(
            overflow in 1usize..8192,
            tail_pieces in proptest::collection::vec(proptest::collection::vec(any::<u8>(), 0..64), 0..8),
        ) {
            let head = vec![b'h'; STREAM_BUFFER_CAP_BYTES + overflow];
            let mut full = head.clone();
            let mut ring = Ring::default();
            ring.append(head);
            for piece in &tail_pieces {
                ring.append(piece.clone());
                full.extend_from_slice(piece);
            }
            prop_assert!(ring.dropped);
            prop_assert_eq!(ring.size, STREAM_BUFFER_CAP_BYTES);
            let tail = ring.tail();
            prop_assert_eq!(tail.len(), STREAM_BUFFER_CAP_BYTES);
            prop_assert!(full.ends_with(&tail));
        }
    }

    proptest! {
        /// A Snapshot assigns the seq basis regardless of prior seqs (the one
        /// deliberate exception to never-rewind), and the ring afterwards holds
        /// exactly the repaint.
        #[test]
        fn pump_snapshot_assigns_basis_and_replaces_ring(
            pre_seqs in proptest::collection::vec(0i64..10_000, 0..16),
            basis in 0i64..10_000,
        ) {
            let shared = test_shared();
            let mut pump = Pump::new(shared.clone());
            for seq in pre_seqs {
                pump.apply_frame(chunk(OutputStream::Stdout, seq, b"x"));
            }
            pump.apply_frame(snapshot_frame(b"repaint", basis));
            prop_assert_eq!(pump.stdout_seq, basis);
            prop_assert_eq!(lock(&shared.state).stdout.tail(), b"repaint".to_vec());
        }
    }

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