sail-rs 0.6.3

//! The Sail client: the canonical async surface that owns configuration and
//! transport, shared by the Python and TypeScript SDKs and the 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 Python SDK, 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 time::OffsetDateTime;

use crate::app::{self, App};
use crate::config::Config;
use crate::credential::api::CredentialApi;
use crate::credential::types::{
    CredentialInjectionPolicyInfo, CredentialInjectionPolicyPage, InjectionRule,
    ListCredentialInjectionPoliciesQuery, SecretInfo,
};
use crate::error::{RpcStatus, SailError};
use crate::exec::{ExecOptions, ExecParams, ExecProcess, ExecResult, OutputStream};
use crate::http::HttpCore;
use crate::imagebuild::BuildMode;
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::{
    is_transient_transport_message, 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,
}

/// Maximum time spent probing a create/resume routing hint before resolving
/// current placement. The relaunch reuses the idempotency key if this expires.
const HINTED_EXEC_ENDPOINT_PROBE_TIMEOUT: Duration = Duration::from_secs(1);

/// 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,
            BuildMode::ReuseExisting,
        );
        let build = 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,
                })
            })?;
        // A create naming a registry tag resolves that tag again server-side,
        // which can name an image the rebuild never produced; the rebuild's
        // pinned reference names exactly what it produced, so the retry
        // creates from that instead of the tag.
        let mut retry = req.clone();
        crate::imagebuild::pin_resolved_oci_ref(&mut retry.image, &build.resolved_oci_ref);
        self.sailbox_api().create(&retry, 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 Sailbox behind under the same name. 10 minutes is a good default;
    /// `None` leaves each attempt unbounded. If the budget is exhausted the
    /// Sailbox 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> {
        // Reject an ambiguous or malformed image source before the VM exists.
        // A directly constructed spec (e.g. from a binding caller) can set both
        // source arms or carry an unvalidated OCI reference, which the backend
        // rejects at request time.
        crate::imagebuild::validate_image_spec_source(&req.image)?;
        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, optionally scheduling a wall-clock wake first.
    #[doc(hidden)]
    pub async fn sleep_sailbox(
        &self,
        sailbox_id: &str,
        wake_at: Option<OffsetDateTime>,
    ) -> Result<Option<OffsetDateTime>, SailError> {
        self.sailbox_api().sleep(sailbox_id, wake_at).await
    }

    /// Replace when Sail may sleep a Sailbox on its own.
    #[doc(hidden)]
    pub async fn set_sailbox_auto_sleep(
        &self,
        sailbox_id: &str,
        auto_sleep: crate::AutoSleep,
    ) -> Result<(), SailError> {
        self.sailbox_api()
            .set_auto_sleep(sailbox_id, auto_sleep)
            .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
    }

    /// Create a new running Sailbox from a durable checkpoint handle. The new
    /// Sailbox restores the memory saved in the checkpoint as well as the
    /// writable disk, so processes the original was running carry on here, and
    /// it runs independently of the Sailbox that took the checkpoint. Commands
    /// started with [`Sailbox::exec`] stop here, though their writes up to the
    /// checkpoint are kept, and one started with `background` keeps running.
    /// Start the other execs the new Sailbox needs. Sometimes it comes up
    /// cold instead, with the disk intact and nothing running, and a
    /// Sailbox that mounts a volume always does. Volumes are mounted on it at
    /// the same paths as on the original, and they are the same volumes, so
    /// both Sailboxes read and write the same files.
    ///
    /// `name` sets the new Sailbox's display name, and the server derives one
    /// when it is omitted. `timeout` is accepted and ignored: the call blocks
    /// until the restore finishes, so apply your own deadline if you need one.
    /// It must be positive when given.
    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.
    /// Re-exposing a port under the same protocol sets its allowlist to what
    /// you pass, so pass the whole list every time; an empty one clears the
    /// restriction and reopens the port.
    #[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 Sailbox.
    #[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
    /// Sailbox; 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)
    }

    /// Fetch the current organization's custom-domain DNS targets.
    #[doc(hidden)]
    pub async fn custom_domain_dns_targets(&self) -> Result<(String, Option<String>), SailError> {
        self.sailbox_api().custom_domain_dns_targets().await
    }

    /// Attach a custom domain to a Sailbox HTTP listener.
    #[doc(hidden)]
    pub async fn attach_custom_domain(
        &self,
        sailbox_id: &str,
        domain: &str,
        guest_port: u32,
    ) -> Result<crate::sailbox::types::CustomDomainInfo, SailError> {
        self.sailbox_api()
            .attach_custom_domain(sailbox_id, domain, guest_port)
            .await
    }

    /// List the custom domains attached to a Sailbox.
    #[doc(hidden)]
    pub async fn list_custom_domains(
        &self,
        sailbox_id: &str,
    ) -> Result<Vec<crate::sailbox::types::CustomDomainInfo>, SailError> {
        self.sailbox_api().list_custom_domains(sailbox_id).await
    }

    /// Detach a custom domain from a Sailbox.
    #[doc(hidden)]
    pub async fn detach_custom_domain(
        &self,
        sailbox_id: &str,
        domain: &str,
    ) -> Result<(), SailError> {
        self.sailbox_api()
            .detach_custom_domain(sailbox_id, domain)
            .await
    }

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

    // --- secrets and credential injection policies ---
    //
    // Id-forms of the surface documented on [`crate::Credentials`],
    // [`crate::Secret`], and [`crate::CredentialInjectionPolicy`]; the bound
    // objects delegate here, and the language bridges call these directly.

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

    /// Set (create or update) a secret's value; returns its metadata.
    #[doc(hidden)]
    pub async fn set_secret(&self, name: &str, value: &str) -> Result<SecretInfo, SailError> {
        self.credential_api().set_secret(name, value).await
    }

    /// Fetch one secret's metadata (never the value).
    #[doc(hidden)]
    pub async fn get_secret(&self, name: &str) -> Result<SecretInfo, SailError> {
        self.credential_api().get_secret(name).await
    }

    /// List the org's secrets, metadata only.
    #[doc(hidden)]
    pub async fn list_secrets(&self) -> Result<Vec<SecretInfo>, SailError> {
        self.credential_api().list_secrets().await
    }

    /// Delete a secret by name.
    #[doc(hidden)]
    pub async fn delete_secret(&self, name: &str) -> Result<(), SailError> {
        self.credential_api().delete_secret(name).await
    }

    /// Create a credential injection policy.
    #[doc(hidden)]
    pub async fn create_credential_policy(
        &self,
        name: &str,
        rules: &[InjectionRule],
    ) -> Result<CredentialInjectionPolicyInfo, SailError> {
        self.credential_api().create_policy(name, rules).await
    }

    /// Fetch one credential injection policy by id.
    #[doc(hidden)]
    pub async fn get_credential_policy(
        &self,
        policy_id: &str,
    ) -> Result<CredentialInjectionPolicyInfo, SailError> {
        self.credential_api().get_policy(policy_id).await
    }

    /// List credential injection policies.
    #[doc(hidden)]
    pub async fn list_credential_policies(
        &self,
        query: &ListCredentialInjectionPoliciesQuery,
    ) -> Result<CredentialInjectionPolicyPage, SailError> {
        self.credential_api().list_policies(query).await
    }

    /// Rename a credential injection policy.
    #[doc(hidden)]
    pub async fn rename_credential_policy(
        &self,
        policy_id: &str,
        name: &str,
    ) -> Result<CredentialInjectionPolicyInfo, SailError> {
        self.credential_api().rename_policy(policy_id, name).await
    }

    /// Delete a credential injection policy by id.
    #[doc(hidden)]
    pub async fn delete_credential_policy(&self, policy_id: &str) -> Result<(), SailError> {
        self.credential_api().delete_policy(policy_id).await
    }

    /// The policy attached to a Sailbox, or `None`.
    #[doc(hidden)]
    pub async fn sailbox_credential_policy(
        &self,
        sailbox_id: &str,
    ) -> Result<Option<CredentialInjectionPolicyInfo>, SailError> {
        self.credential_api().sailbox_policy(sailbox_id).await
    }

    /// Attach a policy to a Sailbox, replacing any previous one.
    #[doc(hidden)]
    pub async fn set_sailbox_credential_policy(
        &self,
        sailbox_id: &str,
        policy_id: &str,
    ) -> Result<(), SailError> {
        self.credential_api()
            .attach_sailbox_policy(sailbox_id, policy_id)
            .await
    }

    /// Detach a Sailbox's credential policy (idempotent).
    #[doc(hidden)]
    pub async fn clear_sailbox_credential_policy(&self, sailbox_id: &str) -> Result<(), SailError> {
        self.credential_api()
            .detach_sailbox_policy(sailbox_id)
            .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> {
        self.exec_at_endpoint(sailbox_id, None, argv, options).await
    }

    /// Start an exec through a previously returned stable workerproxy endpoint.
    /// Create/resume-born Sailbox objects use this to avoid a redundant resume;
    /// id-only objects pass `None` and retain the normal wake-and-resolve path.
    #[doc(hidden)]
    pub async fn exec_at_endpoint(
        &self,
        sailbox_id: &str,
        exec_endpoint: Option<&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 hinted_endpoint = exec_endpoint.filter(|endpoint| !endpoint.is_empty());
        let exec_endpoint = match hinted_endpoint {
            Some(endpoint) => endpoint.to_string(),
            None => 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(),
            // The clipboard bridge is a pty-session behavior; the guest would
            // ignore it elsewhere, so don't ask.
            forward_clipboard: options.forward_clipboard && options.pty,
        };
        self.start_exec_params_at_endpoint(params, hinted_endpoint.is_some())
            .await
    }

    /// Starts already-encoded exec parameters and safely re-resolves a hinted
    /// endpoint after migration. Bindings use this to share the exact retry and
    /// idempotency semantics of [`Client::exec_at_endpoint`].
    #[doc(hidden)]
    pub async fn start_exec_params_at_endpoint(
        &self,
        mut params: ExecParams,
        endpoint_was_hint: bool,
    ) -> Result<ExecProcess, SailError> {
        if !endpoint_was_hint {
            return ExecProcess::start(self.worker(), params).await;
        }

        // A create/resume handle is authoritative when returned, but the VM
        // may migrate before its caller launches exec. Try the hint once with
        // the normal idempotency key, then resolve fresh on any failure that a
        // stale worker can produce. The resolved attempt receives the caller's
        // full retry budget and safely reattaches if the first worker launched
        // the command but lost its Started response.
        params.ensure_idempotency_key();
        let hinted_start =
            ExecProcess::start_with_initial_retry_timeout(self.worker(), params.clone(), Some(0.0));
        match tokio::time::timeout(HINTED_EXEC_ENDPOINT_PROBE_TIMEOUT, hinted_start).await {
            Ok(Ok(process)) => return Ok(process),
            Ok(Err(err)) if !should_reresolve_hinted_exec_endpoint(&err) => return Err(err),
            Ok(Err(_)) | Err(_) => {}
        }

        // connect_lazy owns the dial in tonic's background channel worker, so
        // dropping the timed-out RPC future above does not cancel a stuck
        // connection. Evict the hint before resolving placement: when the
        // public endpoint is unchanged, the fallback must still dial a fresh
        // channel instead of reusing the one whose probe just timed out.
        self.worker().channels().invalidate(&params.exec_endpoint);
        let endpoint = self.exec_endpoint(&params.sailbox_id).await?;
        params.exec_endpoint = endpoint;
        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,
        options: ExecOptions,
    ) -> Result<ExecProcess, SailError> {
        self.exec_shell_at_endpoint(sailbox_id, None, command, options)
            .await
    }

    /// Shell-command counterpart to [`Client::exec_at_endpoint`].
    #[doc(hidden)]
    pub async fn exec_shell_at_endpoint(
        &self,
        sailbox_id: &str,
        exec_endpoint: Option<&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_at_endpoint(sailbox_id, exec_endpoint, 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 up so the server never enforces a
/// shorter bound than the caller asked for. A zero duration stays zero, which
/// the request builders refuse as a non-positive value.
pub(crate) fn duration_to_whole_seconds(duration: Duration) -> i64 {
    duration.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
        ),
    }
}

/// Whether an exec failure can mean a create/resume endpoint hint went stale.
/// Transport messages relayed as source-less UNKNOWN/INTERNAL statuses need
/// the same treatment as structurally retryable transport failures.
fn should_reresolve_hinted_exec_endpoint(err: &SailError) -> bool {
    err.retryable()
        || matches!(
            err,
            SailError::Terminated { .. } | SailError::HostLost { .. }
        )
        || matches!(
            err,
            SailError::Execution {
                code: RpcStatus::Unknown | RpcStatus::Internal,
                detail,
            } if is_transient_transport_message(detail)
        )
}

#[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);
    }

    #[test]
    fn hinted_exec_reresolves_source_less_transport_statuses() {
        let relayed_transport = SailError::Execution {
            code: RpcStatus::Unknown,
            detail: "error reading server preface: EOF".to_string(),
        };
        assert!(should_reresolve_hinted_exec_endpoint(&relayed_transport));

        let server_verdict = SailError::Execution {
            code: RpcStatus::Unknown,
            detail: "application rejected exec".to_string(),
        };
        assert!(!should_reresolve_hinted_exec_endpoint(&server_verdict));
    }
}