sail/

client.rs

//! The Sail client: the canonical async surface that owns configuration and
//! transport, shared by every binding (Python, TypeScript, CLI).
//!
//! [`Client`] is a cheap-to-clone handle (`Arc` inside, like `reqwest::Client`):
//! clone it freely to share the connection pools and config. Construct it with
//! [`Client::from_env`] or [`Client::builder`].
//!
//! Every method is `async`. Synchronous callers (the PyO3 bridge with the GIL
//! released, the CLI) drive these futures with
//! [`crate::block_on`]; an async host awaits them directly.
//!
//! ```no_run
//! # async fn run() -> Result<(), sail::error::SailError> {
//! use sail::Client;
//!
//! // From the environment (SAIL_API_KEY):
//! let client = Client::from_env()?;
//! let page = client.list_sailboxes(&Default::default()).await?;
//! println!("{} sailboxes", page.items.len());
//!
//! // Or build one explicitly:
//! let client = Client::builder("sk_...").build()?;
//! let app = client.find_app("my-app", /* mint_if_missing */ true).await?;
//! # let _ = (client, app);
//! # Ok(())
//! # }
//! ```

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

use crate::app::{self, App};
use crate::config::Config;
use crate::error::{RpcStatus, SailError};
use crate::exec::{ExecOptions, ExecParams, ExecProcess, ExecResult, OutputStream};
use crate::http::HttpCore;
use crate::imagebuilder::ImageBuilder;
use crate::sailbox::api::{SailboxApi, UpgradeResult};
use crate::sailbox::fs::{DirEntry, EntryType};
use crate::sailbox::object::Sailbox;
use crate::sailbox::types::{
    CreateSailboxRequest, ListSailboxesQuery, SailboxCheckpoint, SailboxHandle, SailboxInfo,
    SailboxMetricsQuery, SailboxMetricsResponse, SailboxPage, SailboxSpendQuery,
    SailboxSpendResponse, VolumeInfo, WhoAmI,
};
use crate::worker::{FileReader, FileWriter, Listener, WorkerProxy, WriteOptions};

/// A configured Sail client. Cheap to clone; shares transport across clones.
#[derive(Clone)]
pub struct Client {
    inner: Arc<Inner>,
}

impl std::fmt::Debug for Client {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Client")
            .field("config", &self.inner.config)
            .finish_non_exhaustive()
    }
}

struct Inner {
    config: Config,
    /// Sailbox-API host: lifecycle, list/get, listeners, volume.
    sailbox_http: HttpCore,
    /// Central public-API host: app find, inference, voyages.
    api_http: HttpCore,
    /// Per-sailbox worker proxy: exec, files, listener reads. Its own `Arc` so
    /// the streaming file/exec methods (which take `&Arc<Self>`) can share it.
    worker: Arc<WorkerProxy>,
    imagebuilder: ImageBuilder,
    /// Successful image-readiness builds, shared by every clone of this
    /// client (see [`crate::imagecache`]).
    image_ready: crate::imagecache::ImageReadyCache,
}

/// Builds a [`Client`] from explicit values, falling back to the default
/// endpoints. Prefer [`Client::from_env`] for the common env-driven case.
///
/// `Debug` redacts the API key, so a logged builder never leaks the
/// credential.
#[derive(Default, Clone)]
pub struct ClientBuilder {
    mode: Option<String>,
    api_key: Option<String>,
    api_url: Option<String>,
    sailbox_api_url: Option<String>,
    imagebuilder_url: Option<String>,
    ingress_url: Option<String>,
    client_label: Option<String>,
}

impl std::fmt::Debug for ClientBuilder {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("ClientBuilder")
            .field(
                "api_key",
                &crate::config::redact_key(self.api_key.as_deref().unwrap_or("")),
            )
            .field("mode", &self.mode)
            .field("api_url", &self.api_url)
            .field("sailbox_api_url", &self.sailbox_api_url)
            .field("imagebuilder_url", &self.imagebuilder_url)
            .field("ingress_url", &self.ingress_url)
            .field("client_label", &self.client_label)
            .finish()
    }
}

impl ClientBuilder {
    /// A builder with the given API key; unset endpoints use the Sail
    /// defaults.
    pub fn new(api_key: impl Into<String>) -> ClientBuilder {
        ClientBuilder {
            api_key: Some(api_key.into()),
            ..ClientBuilder::default()
        }
    }

    /// Select the named environment (`prod`/`dev`/`staging`/`local`), which
    /// picks the endpoint defaults. Unset means prod.
    #[doc(hidden)]
    pub fn mode(mut self, mode: impl Into<String>) -> ClientBuilder {
        self.mode = Some(mode.into());
        self
    }

    /// Override the Sail API URL.
    pub fn api_url(mut self, api_url: impl Into<String>) -> ClientBuilder {
        self.api_url = Some(api_url.into());
        self
    }

    /// Override the sailbox-API URL.
    pub fn sailbox_api_url(mut self, url: impl Into<String>) -> ClientBuilder {
        self.sailbox_api_url = Some(url.into());
        self
    }

    /// Override the image-build endpoint (`host:port`).
    pub fn imagebuilder_url(mut self, url: impl Into<String>) -> ClientBuilder {
        self.imagebuilder_url = Some(url.into());
        self
    }

    /// Override the listener ingress base URL (what `SAILBOX_INGRESS_URL`
    /// sets from the environment), for custom or self-hosted Sailbox stacks.
    pub fn ingress_url(mut self, url: impl Into<String>) -> ClientBuilder {
        self.ingress_url = Some(url.into());
        self
    }

    /// Identify the first-party binding using the shared transport.
    #[doc(hidden)]
    pub fn client_label(mut self, label: impl Into<String>) -> ClientBuilder {
        self.client_label = Some(label.into());
        self
    }

    /// Build the client, resolving any unset endpoint from the defaults.
    pub fn build(self) -> Result<Client, SailError> {
        let api_key = self.api_key.unwrap_or_default();
        let config = Config::resolve(
            self.mode.as_deref(),
            api_key,
            self.api_url,
            self.sailbox_api_url,
            self.imagebuilder_url,
            self.ingress_url,
        )?;
        Client::from_config_with_label(
            config,
            self.client_label
                .as_deref()
                .unwrap_or(crate::http::DEFAULT_CLIENT_LABEL),
        )
    }
}

/// Bound on the transparent image rebuild inside a create retry when the
/// request carries no image-build timeout; matches the default build budget
/// the SDK wrappers document.
const STALE_IMAGE_REBUILD_TIMEOUT: Duration = Duration::from_mins(30);

/// The scheduler's create rejection for an image it cannot resolve as ready.
/// CreateSailbox in backend/internal/sailbox/scheduler stamps the
/// "resolve image:" prefix on that arm; keep them in sync.
fn image_not_ready_conflict(result: &Result<SailboxHandle, SailError>) -> bool {
    matches!(
        result,
        Err(SailError::Creation {
            status: 409,
            message,
            ..
        }) if message.starts_with("resolve image:")
    )
}

impl Client {
    /// Start a [`ClientBuilder`].
    pub fn builder(api_key: impl Into<String>) -> ClientBuilder {
        ClientBuilder::new(api_key)
    }

    /// Build a client from the environment (`SAIL_API_KEY`, …).
    pub fn from_env() -> Result<Client, SailError> {
        Client::from_config(Config::from_env()?)
    }

    /// Build a client from the environment and identify a first-party binding.
    #[doc(hidden)]
    pub fn from_env_with_label(label: &str) -> Result<Client, SailError> {
        Client::from_config_with_label(Config::from_env()?, label)
    }

    /// Build a client from an already-resolved [`Config`].
    pub fn from_config(config: Config) -> Result<Client, SailError> {
        Client::from_config_with_label(config, crate::http::DEFAULT_CLIENT_LABEL)
    }

    fn from_config_with_label(config: Config, client_label: &str) -> Result<Client, SailError> {
        let sailbox_http = HttpCore::new(&config.sailbox_api_url, &config.api_key)?
            .with_client_label(client_label);
        let api_http =
            HttpCore::new(&config.api_url, &config.api_key)?.with_client_label(client_label);
        let worker = Arc::new(WorkerProxy::new(&config.api_key)?);
        let imagebuilder = ImageBuilder::new(&config.imagebuilder_url, &config.api_key)?;
        Ok(Client {
            inner: Arc::new(Inner {
                config,
                sailbox_http,
                api_http,
                worker,
                imagebuilder,
                image_ready: crate::imagecache::ImageReadyCache::new(),
            }),
        })
    }

    pub(crate) fn image_ready_cache(&self) -> &crate::imagecache::ImageReadyCache {
        &self.inner.image_ready
    }

    /// Test hook: shrink the window after which a cached successful image
    /// build is re-verified with the server. Compiled only for tests (this
    /// crate's own and, under `test-fakes`, the integration crate), so it
    /// never widens the published API.
    #[cfg(any(test, feature = "test-fakes"))]
    pub fn set_image_ready_refresh_window(&self, window: std::time::Duration) {
        self.inner.image_ready.set_refresh_window(window);
    }

    /// The resolved configuration.
    pub fn config(&self) -> &Config {
        &self.inner.config
    }

    /// The worker proxy for exec, file copy, and listener reads.
    #[doc(hidden)]
    pub fn worker(&self) -> Arc<WorkerProxy> {
        Arc::clone(&self.inner.worker)
    }

    /// The imagebuilder dispatcher client.
    #[doc(hidden)]
    pub fn imagebuilder(&self) -> &ImageBuilder {
        &self.inner.imagebuilder
    }

    /// The sailbox-API HTTP host (for binding-built requests).
    #[doc(hidden)]
    pub fn sailbox_http(&self) -> &HttpCore {
        &self.inner.sailbox_http
    }

    /// The central public-API HTTP host (for binding-built requests).
    #[doc(hidden)]
    pub fn api_http(&self) -> &HttpCore {
        &self.inner.api_http
    }

    fn sailbox_api(&self) -> SailboxApi<'_> {
        SailboxApi::new(&self.inner.sailbox_http)
    }

    /// Send a create; when the scheduler rejects it because the image is not
    /// ready even though readiness was cached, rebuild once and retry. A
    /// backend deploy can change the canonical image identity behind the same
    /// spec, so a cached "ready" can be stale until the refresh window. The
    /// scheduler resolves the image before it creates any row, so nothing
    /// exists server-side and the retried create is safe. Unrelated create
    /// conflicts (name, idempotency) pass through untouched.
    async fn create_with_image_revalidation(
        &self,
        req: &CreateSailboxRequest,
        timeout: Option<Duration>,
    ) -> Result<SailboxHandle, SailError> {
        let create_started = std::time::Instant::now();
        let result = self.sailbox_api().create(req, timeout).await;
        let custom_image = req.image != crate::image::ImageSpec::default()
            && !crate::imagebuild::is_builtin_base_spec(&req.image);
        if !custom_image || !image_not_ready_conflict(&result) {
            return result;
        }
        if let Ok(spec_hash) = crate::imagebuild::canonical_spec_key(&req.image) {
            // Drop every entry whose build started before this create began:
            // those may carry the identity the server just rejected. A build
            // started after conflict discovery is another stale caller's
            // recovery, joined below rather than clobbered.
            self.image_ready_cache()
                .invalidate_spec_started_before(&spec_hash, create_started);
        }
        // The hard envelope means joining another caller's in-flight rebuild
        // cannot outlive this caller's budget; the recovery marking keeps the
        // rebuild joinable through later stale creates' invalidations.
        let rebuild_timeout = req
            .image_build_timeout
            .unwrap_or(STALE_IMAGE_REBUILD_TIMEOUT);
        let rebuild =
            self.build_spec_ready_cached(&req.image, rebuild_timeout, /* recovery */ true);
        tokio::time::timeout(rebuild_timeout, rebuild)
            .await
            .unwrap_or_else(|_| {
                Err(SailError::Transport {
                    kind: crate::error::TransportKind::Timeout,
                    message: "timed out building the image".to_string(),
                    source: None,
                })
            })?;
        self.sailbox_api().create(req, timeout).await
    }

    // --- sailbox lifecycle ---

    /// Create a Sailbox. `timeout` bounds each attempt of the synchronous
    /// create (which can take minutes server-side); the call retries
    /// with one idempotency key so the backend can dedupe rather than
    /// duplicate, and gives up after roughly `max_attempts * timeout`.
    /// An interrupted or re-invoked create is a new request and may leave a
    /// prior box behind under the same name. 10 minutes is a good default;
    /// `None` leaves each attempt unbounded. If the budget is exhausted the
    /// box may still be coming up server-side; find or terminate it by
    /// `name`.
    pub async fn create_sailbox(
        &self,
        req: &CreateSailboxRequest,
        timeout: Option<Duration>,
    ) -> Result<Sailbox, SailError> {
        let bind = |handle: SailboxHandle| Sailbox::bind(self.clone(), handle);
        if !req.ssh {
            return self
                .create_with_image_revalidation(req, timeout)
                .await
                .map(bind);
        }
        // Validate the full request now, port-22 entries included: they are
        // stripped below (their allowlist applies at the enable_ssh expose),
        // so create's own validation never sees them, and an invalid entry
        // must fail here rather than after the VM exists.
        crate::sailbox::api::validate_ingress_ports(&req.ingress_ports)?;
        // SSH setup is org-scoped: preflight the org CA (created on first use)
        // so a CA outage fails before the VM exists.
        self.org_ssh_ca_public_key().await?;
        // Port 22 belongs to enable_ssh, which exposes it only after verifying
        // the CA-only sshd owns it (never the create request), so a failed
        // setup can't leave port 22 exposed. An explicit port-22 entry
        // contributes just its allowlist, applied at that expose.
        let mut req = req.clone();
        let ssh_allowlist = req
            .ingress_ports
            .iter()
            .find(|port| port.guest_port == 22)
            .map(|port| port.allowlist.clone())
            .unwrap_or_default();
        req.ingress_ports.retain(|port| port.guest_port != 22);
        let handle = self.create_with_image_revalidation(&req, timeout).await?;
        let handle_id = handle.sailbox_id.clone();
        // The VM is already up, so skip the readiness probe (wait: false).
        if let Err(err) = self
            .enable_ssh(
                &handle_id,
                &ssh_allowlist,
                /* wait */ false,
                Duration::ZERO,
            )
            .await
        {
            // The sailbox exists; surface its id so the caller can fetch it to
            // retry enable_ssh or terminate it.
            return Err(SailError::Creation {
                message: format!(
                    "sailbox {handle_id} was created, but SSH setup failed: {err}. Fetch it by \
                     id to retry enable_ssh or terminate it."
                ),
                status: 0,
                body: serde_json::Value::Null,
            });
        }
        Ok(bind(handle))
    }

    /// Fetch a single Sailbox.
    #[doc(hidden)]
    pub async fn get_sailbox(&self, sailbox_id: &str) -> Result<SailboxInfo, SailError> {
        self.sailbox_api().get(sailbox_id).await
    }

    /// Fetch the identity (org, and user when user-scoped) behind the API key.
    #[doc(hidden)]
    pub async fn whoami(&self) -> Result<WhoAmI, SailError> {
        self.sailbox_api().whoami().await
    }

    /// List Sailboxes in the current org.
    pub async fn list_sailboxes(
        &self,
        query: &ListSailboxesQuery,
    ) -> Result<SailboxPage, SailError> {
        self.sailbox_api().list(query).await
    }

    /// Estimate Sailbox spend for the current organization over a time window.
    pub async fn sailbox_spend(
        &self,
        query: &SailboxSpendQuery,
    ) -> Result<SailboxSpendResponse, SailError> {
        self.sailbox_api().spend(query).await
    }

    /// Fetch a Sailbox's resource-usage time series.
    pub async fn sailbox_metrics(
        &self,
        sailbox_id: &str,
        query: &SailboxMetricsQuery,
    ) -> Result<SailboxMetricsResponse, SailError> {
        self.sailbox_api().metrics(sailbox_id, query).await
    }

    /// Terminate a Sailbox (idempotent).
    #[doc(hidden)]
    pub async fn terminate_sailbox(&self, sailbox_id: &str) -> Result<(), SailError> {
        self.sailbox_api().terminate(sailbox_id).await
    }

    /// Pause a Sailbox.
    #[doc(hidden)]
    pub async fn pause_sailbox(&self, sailbox_id: &str) -> Result<(), SailError> {
        self.sailbox_api().pause(sailbox_id).await
    }

    /// Sleep a Sailbox.
    #[doc(hidden)]
    pub async fn sleep_sailbox(&self, sailbox_id: &str) -> Result<(), SailError> {
        self.sailbox_api().sleep(sailbox_id).await
    }

    /// Resume a paused/sleeping Sailbox.
    #[doc(hidden)]
    pub async fn resume_sailbox(&self, sailbox_id: &str) -> Result<SailboxHandle, SailError> {
        self.sailbox_api().resume(sailbox_id).await
    }

    /// Checkpoint a running Sailbox.
    #[doc(hidden)]
    pub async fn checkpoint_sailbox(
        &self,
        sailbox_id: &str,
        name: Option<&str>,
        ttl_seconds: Option<i64>,
    ) -> Result<SailboxCheckpoint, SailError> {
        self.sailbox_api()
            .checkpoint(sailbox_id, name, ttl_seconds)
            .await
    }

    /// Fork a Sailbox into a new child in one call. Id-form of
    /// [`Sailbox::fork`](crate::Sailbox::fork), which documents the contract.
    #[doc(hidden)]
    pub async fn fork_sailbox(
        &self,
        sailbox_id: &str,
        name: Option<&str>,
        timeout: Option<Duration>,
    ) -> Result<Sailbox, SailError> {
        self.sailbox_api()
            .fork(sailbox_id, name, timeout.map(duration_to_whole_seconds))
            .await
            .map(|handle| Sailbox::bind(self.clone(), handle))
    }

    /// Create a new Sailbox from a checkpoint.
    pub async fn create_from_checkpoint(
        &self,
        checkpoint_id: &str,
        name: Option<&str>,
        timeout: Option<Duration>,
    ) -> Result<Sailbox, SailError> {
        self.sailbox_api()
            .from_checkpoint(checkpoint_id, name, timeout.map(duration_to_whole_seconds))
            .await
            .map(|handle| Sailbox::bind(self.clone(), handle))
    }

    /// Upgrade the Sailbox runtime (applies now if running, else at next wake).
    #[doc(hidden)]
    pub async fn upgrade_sailbox(&self, sailbox_id: &str) -> Result<UpgradeResult, SailError> {
        self.sailbox_api().upgrade(sailbox_id).await
    }

    /// Expose a guest port at runtime; returns the add-listener response.
    #[doc(hidden)]
    pub async fn expose_listener(
        &self,
        sailbox_id: &str,
        guest_port: u32,
        protocol: crate::sailbox::types::IngressProtocol,
        allowlist: &[String],
    ) -> Result<Listener, SailError> {
        let mut response = self
            .sailbox_api()
            .expose(sailbox_id, guest_port, protocol, allowlist)
            .await?;
        self.fill_listener_url(sailbox_id, &mut response);
        Ok(response)
    }

    /// Remove a runtime ingress port.
    #[doc(hidden)]
    pub async fn unexpose_listener(
        &self,
        sailbox_id: &str,
        guest_port: u32,
    ) -> Result<(), SailError> {
        self.sailbox_api().unexpose(sailbox_id, guest_port).await
    }

    /// List a Sailbox's ingress listeners without resuming (waking) the box.
    #[doc(hidden)]
    pub async fn list_listeners(
        &self,
        sailbox_id: &str,
    ) -> Result<Vec<crate::worker::Listener>, SailError> {
        let mut listeners = self.sailbox_api().list_listeners(sailbox_id).await?;
        for listener in &mut listeners {
            self.fill_listener_url(sailbox_id, listener);
        }
        Ok(listeners)
    }

    /// Fetch one ingress listener by guest port without resuming (waking) the
    /// box; a missing port is a [`SailError::NotFound`].
    #[doc(hidden)]
    pub async fn get_listener(
        &self,
        sailbox_id: &str,
        guest_port: u32,
    ) -> Result<crate::worker::Listener, SailError> {
        let mut listener = self
            .sailbox_api()
            .get_listener(sailbox_id, guest_port)
            .await?;
        self.fill_listener_url(sailbox_id, &mut listener);
        Ok(listener)
    }

    /// Fill an empty `public_url` on a non-TCP listener with the URL
    /// synthesized from this client's ingress config (the server leaves
    /// listener URLs empty in local/path mode).
    fn fill_listener_url(&self, sailbox_id: &str, listener: &mut crate::worker::Listener) {
        if listener.public_url.is_empty()
            && listener.protocol != crate::sailbox::types::ListenerProtocol::Tcp
        {
            listener.public_url = crate::sailbox::listeners::synthesized_public_url(
                self.config(),
                sailbox_id,
                listener.guest_port,
            );
        }
    }

    /// Ingress-identity headers for this Sailbox.
    #[doc(hidden)]
    pub async fn ingress_auth_headers(
        &self,
        sailbox_id: &str,
    ) -> Result<Vec<(String, String)>, SailError> {
        self.sailbox_api().ingress_auth_headers(sailbox_id).await
    }

    /// The caller org's SSH CA public key (created on first use).
    pub async fn org_ssh_ca_public_key(&self) -> Result<String, SailError> {
        self.sailbox_api().org_ssh_ca_public_key().await
    }

    /// Sign `public_key` into a short-lived org-CA certificate (principal
    /// `root`). `timeout` (seconds) bounds a single no-retry attempt; `None`
    /// retries.
    pub async fn issue_user_cert(
        &self,
        public_key: &str,
        timeout: Option<Duration>,
    ) -> Result<crate::sailbox::types::IssuedUserCert, SailError> {
        self.sailbox_api()
            .issue_user_cert(public_key, timeout.map(|t| t.as_secs_f64()))
            .await
    }

    // --- NFS volumes ---

    /// Look up (optionally minting) an NFS volume by name.
    pub async fn get_volume(
        &self,
        name: &str,
        mint_if_missing: bool,
    ) -> Result<VolumeInfo, SailError> {
        self.sailbox_api().get_volume(name, mint_if_missing).await
    }

    /// List NFS volumes in the current org.
    pub async fn list_volumes(
        &self,
        max_objects: Option<i64>,
    ) -> Result<Vec<VolumeInfo>, SailError> {
        self.sailbox_api().list_volumes(max_objects).await
    }

    /// Delete a volume by id.
    pub async fn delete_volume(
        &self,
        volume_id: &str,
        allow_missing: bool,
    ) -> Result<Option<VolumeInfo>, SailError> {
        self.sailbox_api()
            .delete_volume(volume_id, allow_missing)
            .await
    }

    // --- apps (central API) ---

    /// Find an app by name, optionally minting it.
    pub async fn find_app(&self, name: &str, mint_if_missing: bool) -> Result<App, SailError> {
        app::find_app(&self.inner.api_http, name, mint_if_missing).await
    }

    /// Every app the current org owns, newest first.
    pub async fn list_apps(&self) -> Result<Vec<App>, SailError> {
        app::list_apps(&self.inner.api_http).await
    }

    // --- exec and files (per-sailbox worker proxy) ---

    /// Resolve a Sailbox's current worker-proxy endpoint.
    ///
    /// `resume` wakes a paused/sleeping Sailbox and returns its *current*
    /// endpoint, which is the host worker's address and changes when the Sailbox
    /// migrates (e.g. after preemption). The GET Sailbox API omits this routing
    /// field, so resuming is the only way to learn it, and resolving it fresh per
    /// call avoids ever dialing a stale worker.
    #[doc(hidden)]
    pub async fn exec_endpoint(&self, sailbox_id: &str) -> Result<String, SailError> {
        let handle = self.resume_sailbox(sailbox_id).await?;
        if handle.exec_endpoint.is_empty() {
            return Err(SailError::Internal {
                message: format!("sailbox {sailbox_id} resumed without an exec endpoint"),
            });
        }
        Ok(handle.exec_endpoint)
    }

    /// Id-form of [`Sailbox::exec`](crate::Sailbox::exec), which documents the
    /// contract. Spawns the output pump on the calling task's tokio runtime.
    #[doc(hidden)]
    pub async fn exec(
        &self,
        sailbox_id: &str,
        argv: Vec<String>,
        options: ExecOptions,
    ) -> Result<ExecProcess, SailError> {
        if argv.is_empty() {
            return Err(SailError::InvalidArgument {
                message: "command must be non-empty".to_string(),
            });
        }
        if options.cwd.is_some() || options.background {
            return Err(SailError::InvalidArgument {
                message: "cwd and background require a shell command; use exec_shell or run_shell"
                    .to_string(),
            });
        }
        // Validate and encode the env before resolving the endpoint: it is
        // purely local, so a malformed key must not first wake a paused sailbox.
        let env = crate::exec::encode_env(&options.env)?;
        let exec_endpoint = self.exec_endpoint(sailbox_id).await?;
        let params = ExecParams {
            sailbox_id: sailbox_id.to_string(),
            exec_endpoint,
            argv,
            // The wire is whole seconds where 0 means "no limit", so a set
            // sub-second timeout rounds up to 1s rather than collapsing to 0.
            timeout_seconds: options
                .timeout
                .map_or(0, |d| d.as_secs_f64().ceil().max(1.0) as u32),
            idempotency_key: options.idempotency_key,
            // A pty always feeds keystrokes to the command, so it implies an
            // open stdin regardless of the flag.
            open_stdin: options.open_stdin || options.pty,
            pty: options.pty,
            term: options.term,
            cols: options.cols,
            rows: options.rows,
            env,
            retry_timeout: options.retry_timeout.as_secs_f64(),
            forward_ports: options.forward_ports,
            forward_browser: options.forward_browser,
            extra_metadata: Vec::new(),
        };
        ExecProcess::start(self.worker(), params).await
    }

    /// Run a shell command in a Sailbox via `/bin/sh -lc`, honoring the
    /// `cwd`/`background` conveniences in [`ExecOptions`]. See [`Client::exec`]
    /// for the argv form and the runtime notes.
    #[doc(hidden)]
    pub async fn exec_shell(
        &self,
        sailbox_id: &str,
        command: &str,
        mut options: ExecOptions,
    ) -> Result<ExecProcess, SailError> {
        let argv = crate::exec::shell_argv(command, &options)?;
        // The conveniences are baked into the argv now; clear them so the argv
        // path's guard does not re-reject them.
        options.cwd = None;
        options.background = false;
        self.exec(sailbox_id, argv, options).await
    }

    /// Open a streaming read of a guest file. Resumes (wakes) the Sailbox to
    /// reach it; the returned [`FileReader`] yields chunks until end of file.
    ///
    /// # Runtime
    ///
    /// Spawns the read pump on the calling task's tokio runtime (see
    /// [`crate::worker::WorkerProxy::read_file`]).
    #[doc(hidden)]
    pub async fn read_stream(
        &self,
        sailbox_id: &str,
        remote_path: &str,
    ) -> Result<FileReader, SailError> {
        let endpoint = self.exec_endpoint(sailbox_id).await?;
        Ok(self
            .inner
            .worker
            .read_file(&endpoint, sailbox_id, remote_path))
    }

    /// Read a guest file into memory in one call (convenience over
    /// [`Client::read_stream`], which streams a large file without
    /// buffering it whole).
    #[doc(hidden)]
    pub async fn read_file(
        &self,
        sailbox_id: &str,
        remote_path: &str,
    ) -> Result<Vec<u8>, SailError> {
        let reader = self.read_stream(sailbox_id, remote_path).await?;
        let mut contents = Vec::new();
        while let Some(chunk) = reader.next().await {
            contents.extend_from_slice(&chunk?);
        }
        Ok(contents)
    }

    /// Open a streaming write to a guest file. Resumes (wakes) the Sailbox to
    /// reach it; feed the returned [`FileWriter`] with `write_chunk` and end with
    /// `finish`, so a large source is never buffered whole.
    ///
    /// # Runtime
    ///
    /// Spawns the write RPC on the calling task's tokio runtime (see
    /// [`crate::worker::WorkerProxy::write_file`]).
    #[doc(hidden)]
    pub async fn write_stream(
        &self,
        sailbox_id: &str,
        remote_path: &str,
        options: WriteOptions,
    ) -> Result<FileWriter, SailError> {
        let endpoint = self.exec_endpoint(sailbox_id).await?;
        Ok(self.inner.worker.write_file(
            &endpoint,
            sailbox_id,
            remote_path,
            options.create_parents,
            options.mode,
        ))
    }

    /// Write `data` to a guest file in one call (convenience over
    /// [`Client::write_stream`], which streams a large source without
    /// buffering it whole).
    #[doc(hidden)]
    pub async fn write_file(
        &self,
        sailbox_id: &str,
        remote_path: &str,
        data: &[u8],
        options: WriteOptions,
    ) -> Result<(), SailError> {
        let mut writer = self.write_stream(sailbox_id, remote_path, options).await?;
        writer.write(data).await?;
        writer.finish().await
    }

    // --- filesystem helpers ---
    //
    // These build a coreutils command, run it to completion, and inspect the
    // result. They live in the core so the command construction and the `find`
    // output parse are defined once, and every language binding consumes the
    // structured results rather than re-parsing `find`'s output.

    /// Run a command to completion and return its buffered result.
    async fn run_argv(&self, sailbox_id: &str, argv: Vec<String>) -> Result<ExecResult, SailError> {
        self.exec(sailbox_id, argv, ExecOptions::default())
            .await?
            .wait()
            .await
    }

    /// Create a directory and any missing parents (like `mkdir -p`); a no-op if
    /// it already exists.
    #[doc(hidden)]
    pub async fn make_dir(&self, sailbox_id: &str, path: &str) -> Result<(), SailError> {
        crate::sailbox::fs::require_path(path)?;
        let result = self
            .run_argv(
                sailbox_id,
                vec![
                    "mkdir".to_string(),
                    "-p".to_string(),
                    "--".to_string(),
                    path.to_string(),
                ],
            )
            .await?;
        fs_command_ok(&result, &format!("create directory {path}"))
    }

    /// Remove a file or directory tree (like `rm -rf`); a no-op if it is already
    /// absent.
    #[doc(hidden)]
    pub async fn remove_path(&self, sailbox_id: &str, path: &str) -> Result<(), SailError> {
        crate::sailbox::fs::require_path(path)?;
        let result = self
            .run_argv(
                sailbox_id,
                vec![
                    "rm".to_string(),
                    "-rf".to_string(),
                    "--".to_string(),
                    path.to_string(),
                ],
            )
            .await?;
        fs_command_ok(&result, &format!("remove {path}"))
    }

    /// Whether `path` exists in the guest. Follows symlinks (like `test -e`), so
    /// a dangling symlink reports `false`.
    #[doc(hidden)]
    pub async fn path_exists(&self, sailbox_id: &str, path: &str) -> Result<bool, SailError> {
        crate::sailbox::fs::require_path(path)?;
        let result = self
            .run_argv(
                sailbox_id,
                vec!["test".to_string(), "-e".to_string(), path.to_string()],
            )
            .await?;
        // `test -e` answers with its exit code: 0 exists, 1 does not. Any other
        // code (for example a signal-killed process) is a failed check, not an
        // answer, so surface it rather than reading it as absent.
        match result.exit_code {
            0 => Ok(true),
            1 => Ok(false),
            _ => Err(fs_command_error(
                &result,
                &format!("check whether {path} exists"),
            )),
        }
    }

    /// List a directory's immediate entries (files and subdirectories, no
    /// recursion). Requires GNU `find`, which the default Debian image ships. A
    /// missing path errors, as does a path that exists but is not a directory.
    #[doc(hidden)]
    pub async fn list_dir(&self, sailbox_id: &str, path: &str) -> Result<Vec<DirEntry>, SailError> {
        crate::sailbox::fs::require_path(path)?;
        let process = self
            .exec(
                sailbox_id,
                crate::sailbox::fs::list_dir_argv(path),
                ExecOptions::default(),
            )
            .await?;
        let result = process.wait().await?;
        fs_command_ok(&result, &format!("list directory {path}"))?;
        // Buffered stdout is a capped, drop-oldest tail, so parsing a truncated
        // listing would silently drop entries.
        if result.stdout_truncated {
            return Err(SailError::Execution {
                code: RpcStatus::FailedPrecondition,
                detail: format!(
                    "directory listing for {path} was truncated because it has \
                     too many entries; list a smaller subtree"
                ),
            });
        }
        // The records are NUL-terminated, and only the raw buffered bytes keep
        // NUL: the string-typed `ExecResult` replaces it, as does the persisted
        // tail that `wait` falls back to when the live stream loses its ending.
        // So parse the local raw bytes, and require that the stream delivered
        // them all; when it did not, the local buffer may be missing entries.
        if !result.stdout_complete {
            return Err(SailError::Execution {
                code: RpcStatus::FailedPrecondition,
                detail: format!(
                    "directory listing for {path} was interrupted before it \
                     finished streaming; retry the call"
                ),
            });
        }
        let mut entries =
            crate::sailbox::fs::parse_dir_entries(&process.buffered_output(OutputStream::Stdout))
                .map_err(|detail| SailError::Execution {
                code: RpcStatus::FailedPrecondition,
                detail: format!("directory listing for {path} could not be used: {detail}"),
            })?;
        // `find` emits the start point itself as the first record, carrying the
        // path's own type.
        if entries.is_empty() {
            return Err(SailError::Execution {
                code: RpcStatus::FailedPrecondition,
                detail: format!(
                    "directory listing for {path} produced no records; \
                     listing requires GNU find in the guest"
                ),
            });
        }
        let start = entries.remove(0);
        if start.entry_type != EntryType::Directory {
            return Err(SailError::Execution {
                code: RpcStatus::FailedPrecondition,
                detail: format!(
                    "{path} is not a directory (it is a {})",
                    start.entry_type.as_str()
                ),
            });
        }
        Ok(entries)
    }
}

/// Whole seconds for the wire, rounding a positive duration up (like the
/// exec timeout) so the server never enforces a shorter bound than the
/// caller asked for; an explicit zero stays zero for the API to reject.
fn duration_to_whole_seconds(timeout: Duration) -> i64 {
    if timeout.is_zero() {
        0
    } else {
        timeout.as_secs_f64().ceil() as i64
    }
}

/// Fail on a non-zero exit from a filesystem helper command.
fn fs_command_ok(result: &ExecResult, action: &str) -> Result<(), SailError> {
    if result.exit_code != 0 {
        return Err(fs_command_error(result, action));
    }
    Ok(())
}

/// The error for a failed filesystem helper command, folding the guest's stderr
/// into the message.
fn fs_command_error(result: &ExecResult, action: &str) -> SailError {
    let stderr = result.stderr.trim();
    let suffix = if stderr.is_empty() {
        String::new()
    } else {
        format!(": {stderr}")
    };
    SailError::Execution {
        code: RpcStatus::FailedPrecondition,
        detail: format!(
            "failed to {action} (exit code {}){suffix}",
            result.exit_code
        ),
    }
}

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

    #[test]
    fn durations_round_up_to_whole_seconds() {
        assert_eq!(duration_to_whole_seconds(Duration::ZERO), 0);
        assert_eq!(duration_to_whole_seconds(Duration::from_millis(500)), 1);
        assert_eq!(duration_to_whole_seconds(Duration::from_millis(1500)), 2);
        assert_eq!(duration_to_whole_seconds(Duration::from_mins(1)), 60);
    }
}