sail-rs 0.2.11

//! Typed client over the sailbox HTTP API (lifecycle, listeners, and volumes).
//!
//! Each method builds the request body, runs the HTTP call (with retry +
//! Idempotency-Key from [`HttpCore`]), maps a non-2xx response onto the
//! canonical [`SailError`] taxonomy, validates the echoed status, and
//! deserializes the body into a typed struct.

use std::time::Duration;

use serde_json::{json, Value};
use time::format_description::well_known::Rfc3339;
use time::OffsetDateTime;

use crate::error::SailError;
use crate::http::{HttpCore, IdempotencyKey, Method, RequestSpec};
use crate::retry::{RetryPolicy, DEFAULT_RETRY_POLICY, NO_RETRY};
use crate::sailbox::types::{
    AddListenerWire, CreateSailboxRequest, IngressPort, IngressProtocol, IssuedUserCert,
    ListSailboxesQuery, SailboxCheckpoint, SailboxHandle, SailboxInfo, SailboxPage, SailboxStatus,
    VolumeInfo, VolumeMount,
};
use crate::worker::Listener;

// Platform-reserved guest ports, per protocol. 22 = ssh; 10000/10001 = the guest
// agent's control/streaming listeners; 15001/15002 = the in-guest tcp-sidecar.
// HTTP ingress reserves ssh plus the infra ports; TCP intentionally allows 22
// (exposing sshd is the canonical use) but still reserves the rest.
const RESERVED_HTTP_INGRESS_PORTS: &[u32] = &[22, 10000, 10001, 15001, 15002];
const RESERVED_TCP_INGRESS_PORTS: &[u32] = &[10000, 10001, 15001, 15002];
// Well-known unauthenticated service ports (databases/caches) that must not be
// exposed as raw public TCP without a source allowlist.
const UNAUTHENTICATED_TCP_SERVICE_PORTS: &[u32] = &[5432, 3306, 6379, 27017, 9200, 11211];
// Guest runtime paths a volume mount must not cover.
const VOLUME_RESERVED_MOUNT_PATHS: &[&str] =
    &["/dev", "/proc", "/sys", "/run/sail", "/var/run/sail"];

/// Whether an allowlist entry is an IP address or CIDR prefix (as opposed to an
/// app-name entry, which is HTTP-only).
fn is_cidr_or_ip(value: &str) -> bool {
    value.parse::<ipnet::IpNet>().is_ok() || value.parse::<std::net::IpAddr>().is_ok()
}

/// Per-size [floor, max] memory/disk ceilings in whole GiB. The scheduler is
/// the authoritative ladder and re-validates at create; this fails fast
/// before an image build. Locked to the scheduler by the backend's
/// TestSDKSizesMatchScheduler.
const SAILBOX_SIZE_RANGES: &[(&str, u32, u32, u32, u32)] =
    &[("s", 2, 64, 8, 128), ("m", 8, 128, 32, 512)];

/// The size whose range applies when a create names no size.
const DEFAULT_SIZE_LABEL: &str = "m";

/// Validate optional create-time memory/disk ceilings (whole GiB) against the
/// named size's range (the default size's range when unsized).
pub fn validate_size_limits(
    size: Option<&str>,
    memory_gib: Option<u32>,
    disk_gib: Option<u32>,
) -> Result<(), SailError> {
    let label = size.unwrap_or(DEFAULT_SIZE_LABEL);
    let Some((_, mem_min, mem_max, disk_min, disk_max)) = SAILBOX_SIZE_RANGES
        .iter()
        .find(|(name, ..)| *name == label)
        .copied()
    else {
        // An unknown size is rejected by the size parser before this runs.
        return Ok(());
    };
    if let Some(memory_gib) = memory_gib {
        if !(mem_min..=mem_max).contains(&memory_gib) {
            return Err(SailError::InvalidArgument {
                message: format!(
                    "memory_gib for size {label} must be between {mem_min} and {mem_max}"
                ),
            });
        }
    }
    if let Some(disk_gib) = disk_gib {
        if !(disk_min..=disk_max).contains(&disk_gib) {
            return Err(SailError::InvalidArgument {
                message: format!(
                    "disk_gib for size {label} must be between {disk_min} and {disk_max}"
                ),
            });
        }
    }
    Ok(())
}

/// Validate the autosleep timeout: when set, whole minutes in 1..=1440. The
/// scheduler enforces the same range; `None` disables autosleep.
pub fn validate_autosleep_timeout(minutes: Option<i64>) -> Result<(), SailError> {
    if let Some(minutes) = minutes {
        if !(1..=1440).contains(&minutes) {
            return Err(SailError::InvalidArgument {
                message: "autosleep_timeout_minutes must be between 1 and 1440".to_string(),
            });
        }
    }
    Ok(())
}

/// Normalize an allowlist for the wire: trim entries, canonicalize CIDRs
/// (clearing host bits, expanding a bare IP to its /32 or /128), reject a
/// slash-containing entry that is not a valid CIDR (app names never contain a
/// slash, so it can only be a typo), and dedupe. App-name entries pass through
/// verbatim.
pub fn normalize_allowlist(entries: &[String]) -> Result<Vec<String>, SailError> {
    let mut out = Vec::new();
    let mut seen = std::collections::HashSet::new();
    for entry in entries {
        let value = entry.trim();
        if value.is_empty() {
            continue;
        }
        let normalized = if let Ok(net) = value.parse::<ipnet::IpNet>() {
            net.trunc().to_string()
        } else if let Ok(addr) = value.parse::<std::net::IpAddr>() {
            match addr {
                std::net::IpAddr::V4(v4) => format!("{v4}/32"),
                std::net::IpAddr::V6(v6) => format!("{v6}/128"),
            }
        } else if value.contains('/') {
            return Err(SailError::InvalidArgument {
                message: format!("allowlist entry {value:?} is not a valid CIDR"),
            });
        } else {
            value.to_string()
        };
        if seen.insert(normalized.clone()) {
            out.push(normalized);
        }
    }
    Ok(out)
}

/// Validate ingress ports against the platform's per-protocol rules (reserved
/// ports, 1..65535 range, unique guest port, allowlist shape). The scheduler
/// enforces the same rules; this is the shared client-side source of truth so a
/// wrapper fails fast before an image build. An empty list is accepted.
pub fn validate_ingress_ports(ports: &[IngressPort]) -> Result<(), SailError> {
    let invalid = |message: String| Err(SailError::InvalidArgument { message });
    let mut seen = std::collections::HashSet::new();
    for port in ports {
        if port.guest_port < 1 || port.guest_port > 65535 {
            return invalid("ingress_ports must be between 1 and 65535".to_string());
        }
        let is_tcp = matches!(port.protocol, IngressProtocol::Tcp);
        let mut has_allowlist = false;
        for entry in &port.allowlist {
            let value = entry.trim();
            if value.is_empty() {
                continue;
            }
            has_allowlist = true;
            let is_cidr = is_cidr_or_ip(value);
            // A '/' can only be intended as a CIDR (app names never contain one).
            if value.contains('/') && !is_cidr {
                return invalid(format!("allowlist entry {value:?} is not a valid CIDR"));
            }
            // CR-soon nbaruah: allow app-name entries on tcp once the
            // tcp-spooler short-circuit gives raw-TCP connections a source app
            // identity (see workerproxy/tcp_ingress.go). Until then the public
            // TCP relay carries no app identity, so an app-name entry on a tcp
            // listener can never match and would mint a listener that denies
            // every real connection.
            if is_tcp && !is_cidr {
                return invalid(format!(
                    "allowlist entry {value:?}: app-name entries are not supported for tcp \
                     listeners; tcp allowlists must be CIDR prefixes"
                ));
            }
        }
        let reserved = if is_tcp {
            RESERVED_TCP_INGRESS_PORTS
        } else {
            RESERVED_HTTP_INGRESS_PORTS
        };
        if reserved.contains(&port.guest_port) {
            if port.guest_port == 22 {
                return invalid(
                    "guest port 22 is reserved for ssh and cannot be exposed as an http port; \
                     expose it as raw TCP instead"
                        .to_string(),
                );
            }
            if port.guest_port == 10000 || port.guest_port == 10001 {
                return invalid(format!(
                    "guest port {} is reserved for the in-guest sail agent and cannot be exposed \
                     as an ingress port",
                    port.guest_port
                ));
            }
            let proto = if is_tcp { "tcp" } else { "http" };
            return invalid(format!(
                "ingress_ports contains reserved {proto} port {}",
                port.guest_port
            ));
        }
        if !seen.insert(port.guest_port) {
            return invalid(format!(
                "ingress_ports guest port {} must be unique",
                port.guest_port
            ));
        }
        if is_tcp && UNAUTHENTICATED_TCP_SERVICE_PORTS.contains(&port.guest_port) && !has_allowlist
        {
            return invalid(format!(
                "guest port {} is a well-known unauthenticated service port; exposing it as raw \
                 public TCP with no source restriction is a common breach vector. Pass an \
                 allowlist of CIDR prefixes to restrict source IPs (use [\"0.0.0.0/0\", \"::/0\"] \
                 to allow every source)",
                port.guest_port
            ));
        }
    }
    Ok(())
}

/// Normalize a POSIX path like Python's `posixpath.normpath`: collapse repeated
/// slashes, resolve `.`/`..`, and drop a trailing slash, preserving a leading
/// `/`. Used to compare volume mount paths for overlap.
fn normalize_posix_path(path: &str) -> String {
    let is_absolute = path.starts_with('/');
    let mut parts: Vec<&str> = Vec::new();
    for segment in path.split('/') {
        match segment {
            "" | "." => {}
            ".." => {
                if parts.last().is_some_and(|&last| last != "..") {
                    parts.pop();
                } else if !is_absolute {
                    parts.push("..");
                }
            }
            other => parts.push(other),
        }
    }
    let joined = parts.join("/");
    if is_absolute {
        format!("/{joined}")
    } else if joined.is_empty() {
        ".".to_string()
    } else {
        joined
    }
}

/// Whether path `a` equals, contains, or is contained by path `b`.
fn mount_paths_overlap(a: &str, b: &str) -> bool {
    a == b || a.starts_with(&format!("{b}/")) || b.starts_with(&format!("{a}/"))
}

/// Validate volume mounts: each path must be absolute, not the filesystem root,
/// not cover a reserved guest runtime path, and not overlap another mount; each
/// volume id must be non-empty. Shared client-side source of truth.
pub fn validate_volume_mounts(mounts: &[VolumeMount]) -> Result<(), SailError> {
    let invalid = |message: String| Err(SailError::InvalidArgument { message });
    let mut seen: Vec<String> = Vec::new();
    for mount in mounts {
        if mount.volume_id.trim().is_empty() {
            return invalid("volume mount volume_id must not be empty".to_string());
        }
        let path = normalize_posix_path(mount.mount_path.trim());
        if !path.starts_with('/') {
            return invalid("volume mount paths must be absolute".to_string());
        }
        if path == "/" {
            return invalid("volume mount path must not be filesystem root".to_string());
        }
        if VOLUME_RESERVED_MOUNT_PATHS
            .iter()
            .any(|reserved| mount_paths_overlap(&path, reserved))
        {
            return invalid(format!("volume mount path {path:?} is reserved"));
        }
        if seen.iter().any(|other| mount_paths_overlap(&path, other)) {
            return invalid("volume mount paths must not overlap".to_string());
        }
        seen.push(path);
    }
    Ok(())
}

/// Result of an in-guest agent upgrade.
#[derive(Debug, Clone, serde::Serialize)]
pub struct UpgradeResult {
    /// True when applied immediately (running box); false when deferred to the
    /// next wake.
    pub applied: bool,
    /// Lifecycle status of the sailbox after the upgrade call.
    pub status: SailboxStatus,
}

/// Typed client over the sailbox HTTP API host.
pub struct SailboxApi<'a> {
    http: &'a HttpCore,
}

impl<'a> SailboxApi<'a> {
    /// Build a client bound to a sailbox-API `HttpCore`.
    pub fn new(http: &'a HttpCore) -> SailboxApi<'a> {
        SailboxApi { http }
    }

    /// Create a sailbox (`POST /v1/sailboxes`) and wait for it to come up.
    ///
    /// The resource size is included only when set.
    /// A non-2xx response or an echoed `failed` status maps to
    /// [`SailError::Creation`]; the returned handle requires a `running` status.
    ///
    /// `timeout` bounds each attempt: this is a synchronous long-poll the
    /// scheduler can block on for many minutes (capacity queue + VM boot), so
    /// without a bound the call can hang. Retries within one call reuse the
    /// request's idempotency key so the backend can dedupe them; the call
    /// returns as soon as the box is ready and gives up after the policy's
    /// attempts, roughly `max_attempts * timeout`. That dedupe is
    /// best-effort — a re-invoked create is a new request, and an attempt the
    /// scheduler failed for capacity may not be resumable — so an interrupted
    /// create can leave a failed box behind under the same name. `None` leaves each attempt unbounded. If
    /// the budget is exhausted the box may still be coming up server-side; the
    /// caller never saw its id, so find or terminate it by `name`.
    pub async fn create(
        &self,
        req: &CreateSailboxRequest,
        timeout: Option<Duration>,
    ) -> Result<SailboxHandle, SailError> {
        validate_size_limits(req.size.map(|s| s.as_str()), req.memory_gib, req.disk_gib)?;
        validate_ingress_ports(&req.ingress_ports)?;
        validate_autosleep_timeout(req.autosleep_timeout_minutes)?;
        validate_volume_mounts(&req.volume_mounts)?;
        let mut ingress_ports = req.ingress_ports.clone();
        for port in &mut ingress_ports {
            port.allowlist = normalize_allowlist(&port.allowlist)?;
        }
        // Serialize the typed ImageSpec to canonical proto-JSON (camelCase field
        // names and enum value names, via its serde attributes): an ordinary JSON
        // object the backend decodes with protojson, no shape-matching by hand.
        let image = serde_json::to_value(&req.image).map_err(|e| SailError::Internal {
            message: format!("failed to serialize image spec: {e}"),
        })?;
        let mut body = json!({
            "app_id": req.app_id,
            "name": req.name,
            "ingress_ports": ingress_ports,
            "volume_mounts": req.volume_mounts,
            "image": image,
        });
        if let Some(size) = req.size {
            body["size"] = json!(size.as_str());
        }
        if let Some(memory_gib) = req.memory_gib {
            body["memory_limit_gib"] = json!(memory_gib);
        }
        if let Some(disk_gib) = req.disk_gib {
            body["state_disk_limit_gib"] = json!(disk_gib);
        }
        if let Some(timeout) = req.autosleep_timeout_minutes {
            body["autosleep_timeout_min"] = json!(timeout);
        }
        let bytes = serde_json::to_vec(&body).map_err(|e| SailError::Internal {
            message: format!("failed to serialize request body: {e}"),
        })?;
        let (status, data) = self
            .request(
                Method::Post,
                "/v1/sailboxes",
                &[],
                Some(bytes),
                DEFAULT_RETRY_POLICY,
                timeout.map(|d| d.as_secs_f64()),
            )
            .await?;
        // create's documented failure type is Creation; only auth stays
        // PermissionDenied.
        raise_for_create_status(status, &data)?;
        if status_of(&data) == SailboxStatus::Failed {
            return Err(SailError::Creation {
                message: format!(
                    "Sailbox creation failed: {}",
                    data.get("error_message")
                        .and_then(Value::as_str)
                        .unwrap_or("")
                ),
                status,
                body: data,
            });
        }
        require_status(&data, SailboxStatus::Running, status).map_err(|_| SailError::Creation {
            message: format!(
                "Sailbox creation returned unexpected status: {}",
                data.get("status").and_then(Value::as_str).unwrap_or("")
            ),
            status,
            body: data.clone(),
        })?;
        Ok(handle_from(&data, &req.name))
    }

    /// Fetch the current state of one sailbox (`GET /v1/sailboxes/{id}`).
    pub async fn get(&self, sailbox_id: &str) -> Result<SailboxInfo, SailError> {
        let (status, data) = self
            .get_request(&format!("/v1/sailboxes/{sailbox_id}"), &[])
            .await?;
        raise_api_error(status, &data, &format!("Sailbox {sailbox_id:?}"))?;
        info_from(data)
    }

    /// List sailboxes (`GET /v1/sailboxes`) with paging and optional filters.
    ///
    /// App, status, search, and max-guest-schema-version filters are sent only
    /// when present; returns one page plus the paging cursor fields.
    pub async fn list(&self, query: &ListSailboxesQuery) -> Result<SailboxPage, SailError> {
        let mut params: Vec<(String, String)> = vec![
            ("limit".to_string(), query.limit.to_string()),
            ("offset".to_string(), query.offset.to_string()),
        ];
        if let Some(app) = &query.app_id {
            params.push(("app".to_string(), app.clone()));
        }
        if let Some(s) = &query.status {
            params.push(("status".to_string(), s.as_str().to_string()));
        }
        if let Some(s) = &query.search {
            params.push(("search".to_string(), s.clone()));
        }
        if let Some(v) = query.max_guest_schema_version {
            params.push(("max_guest_schema_version".to_string(), v.to_string()));
        }
        let (status, data) = self.get_request("/v1/sailboxes", &params).await?;
        raise_api_error(status, &data, "")?;
        let items = data
            .get("data")
            .and_then(Value::as_array)
            .ok_or_else(|| missing_field("data"))?
            .iter()
            .cloned()
            .map(info_from)
            .collect::<Result<Vec<_>, _>>()?;
        // Tolerate a missing pagination counter: echo the request's limit/offset
        // and fall back to the page size for total, so one absent field does not
        // sink the whole list.
        let total = int_field(&data, "total").unwrap_or(items.len() as i64);
        Ok(SailboxPage {
            limit: int_field(&data, "limit").unwrap_or(query.limit),
            offset: int_field(&data, "offset").unwrap_or(query.offset),
            total,
            has_more: data
                .get("has_more")
                .and_then(Value::as_bool)
                .unwrap_or(false),
            items,
        })
    }

    /// Idempotent: a resource-miss 404 is the same outcome as a fresh terminate.
    pub async fn terminate(&self, sailbox_id: &str) -> Result<(), SailError> {
        let (status, data) = self
            .post(&format!("/v1/sailboxes/{sailbox_id}/terminate"), &json!({}))
            .await?;
        if status == 404 && is_resource_not_found(&data) {
            return Ok(());
        }
        raise_api_error(status, &data, "")
    }

    /// Pause a sailbox (`POST /v1/sailboxes/{id}/pause`), keeping it in memory.
    ///
    /// Requires the echoed status to be `paused`.
    pub async fn pause(&self, sailbox_id: &str) -> Result<(), SailError> {
        self.stop(sailbox_id, "pause", SailboxStatus::Paused).await
    }

    /// Sleep a sailbox (`POST /v1/sailboxes/{id}/sleep`), persisting it to disk.
    ///
    /// Requires the echoed status to be `sleeping`.
    pub async fn sleep(&self, sailbox_id: &str) -> Result<(), SailError> {
        self.stop(sailbox_id, "sleep", SailboxStatus::Sleeping)
            .await
    }

    async fn stop(
        &self,
        sailbox_id: &str,
        action: &str,
        expected: SailboxStatus,
    ) -> Result<(), SailError> {
        let (status, data) = self
            .post(&format!("/v1/sailboxes/{sailbox_id}/{action}"), &json!({}))
            .await?;
        raise_api_error(status, &data, "")?;
        require_status(&data, expected, status)
    }

    /// Resume a paused or sleeping sailbox (`POST /v1/sailboxes/{id}/resume`).
    ///
    /// A `terminal_unavailable` resume state maps to [`SailError::NotFound`]
    /// since a terminated sailbox cannot be brought back; otherwise requires a
    /// `running` status and returns a fresh handle.
    pub async fn resume(&self, sailbox_id: &str) -> Result<SailboxHandle, SailError> {
        let (status, data) = self
            .post(&format!("/v1/sailboxes/{sailbox_id}/resume"), &json!({}))
            .await?;
        raise_api_error(status, &data, "")?;
        if data.get("resume_state").and_then(Value::as_str) == Some("terminal_unavailable") {
            // A terminated sailbox is gone for good; NotFound maps to the
            // binding's lookup error.
            return Err(SailError::NotFound {
                message: data
                    .get("error_message")
                    .and_then(Value::as_str)
                    .filter(|s| !s.is_empty())
                    .unwrap_or("sailbox cannot be resumed")
                    .to_string(),
            });
        }
        require_status(&data, SailboxStatus::Running, status)?;
        Ok(handle_from(&data, ""))
    }

    /// Checkpoint a running sailbox (`POST /v1/sailboxes/{id}/checkpoint`).
    ///
    /// `name` sets the handle's display name; `ttl_seconds`, when given, must be
    /// positive and overrides the server's default retention. Returns the new
    /// checkpoint's id, generation, and expiry; a response missing
    /// `checkpoint_id` is treated as an [`SailError::Api`] failure.
    pub async fn checkpoint(
        &self,
        sailbox_id: &str,
        name: Option<&str>,
        ttl_seconds: Option<i64>,
    ) -> Result<SailboxCheckpoint, SailError> {
        let mut body = json!({});
        if let Some(name) = name {
            body["name"] = json!(name);
        }
        if let Some(ttl) = ttl_seconds {
            if ttl <= 0 {
                return Err(SailError::InvalidArgument {
                    message: "ttl_seconds must be greater than 0".to_string(),
                });
            }
            body["ttl_seconds"] = json!(ttl);
        }
        let (status, data) = self
            .post(&format!("/v1/sailboxes/{sailbox_id}/checkpoint"), &body)
            .await?;
        raise_api_error(status, &data, "")?;
        let checkpoint_id = data
            .get("checkpoint_id")
            .and_then(Value::as_str)
            .filter(|s| !s.is_empty());
        let checkpoint_id = if let Some(id) = checkpoint_id {
            id.to_string()
        } else {
            require_status(&data, SailboxStatus::Running, status)?;
            return Err(SailError::Api {
                status,
                message: "checkpoint sailbox did not return checkpoint_id".to_string(),
                body: data,
            });
        };
        Ok(SailboxCheckpoint {
            checkpoint_id,
            sailbox_id: data
                .get("sailbox_id")
                .and_then(Value::as_str)
                .unwrap_or(sailbox_id)
                .to_string(),
            checkpoint_generation: data
                .get("checkpoint_generation")
                .and_then(Value::as_i64)
                .unwrap_or(0),
            expires_at: data
                .get("expires_at")
                .and_then(Value::as_str)
                .and_then(|s| OffsetDateTime::parse(s, &Rfc3339).ok()),
            status: status_of(&data),
        })
    }

    /// Create a sailbox from an existing checkpoint
    /// (`POST /v1/sailboxes/from_checkpoint`).
    ///
    /// `checkpoint_id` must be non-empty and `timeout_seconds`, when given, must
    /// be positive; both are validated as [`SailError::InvalidArgument`] before
    /// the call. Requires a `running` status and returns the new sailbox handle.
    pub async fn from_checkpoint(
        &self,
        checkpoint_id: &str,
        name: Option<&str>,
        timeout_seconds: Option<i64>,
    ) -> Result<SailboxHandle, SailError> {
        if checkpoint_id.is_empty() {
            return Err(SailError::InvalidArgument {
                message: "checkpoint_id is required".to_string(),
            });
        }
        let mut body = json!({ "checkpoint_id": checkpoint_id });
        if let Some(name) = name {
            body["name"] = json!(name);
        }
        if let Some(timeout) = timeout_seconds {
            if timeout <= 0 {
                return Err(SailError::InvalidArgument {
                    message: "timeout must be greater than 0".to_string(),
                });
            }
            body["timeout_seconds"] = json!(timeout);
        }
        let (status, data) = self.post("/v1/sailboxes/from_checkpoint", &body).await?;
        raise_api_error(status, &data, "")?;
        require_status(&data, SailboxStatus::Running, status)?;
        Ok(handle_from(&data, name.unwrap_or("")))
    }

    /// Upgrade a sailbox's in-guest agent (`POST /v1/sailboxes/{id}/upgrade`).
    ///
    /// The returned [`UpgradeResult`] reports whether the upgrade applied
    /// immediately (running box) or was deferred to the next wake.
    pub async fn upgrade(&self, sailbox_id: &str) -> Result<UpgradeResult, SailError> {
        let (status, data) = self
            .post(&format!("/v1/sailboxes/{sailbox_id}/upgrade"), &json!({}))
            .await?;
        raise_api_error(status, &data, "")?;
        Ok(UpgradeResult {
            applied: data
                .get("applied")
                .and_then(Value::as_bool)
                .unwrap_or(false),
            status: status_of(&data),
        })
    }

    /// Expose a runtime ingress port; returns the [`Listener`](crate::worker::Listener)
    /// with the endpoint the scheduler resolved and an unspecified route status
    /// (the response does not report reachability).
    pub async fn expose(
        &self,
        sailbox_id: &str,
        guest_port: u32,
        protocol: IngressProtocol,
        allowlist: &[String],
    ) -> Result<crate::worker::Listener, SailError> {
        // Always send allowlist (even empty) so re-exposing a port can clear it
        // and the request is explicit rather than relying on absence semantics.
        let allowlist = normalize_allowlist(allowlist)?;
        let body = json!({
            "guest_port": guest_port,
            "protocol": protocol.as_str(),
            "allowlist": allowlist,
        });
        let (status, data) = self
            .post(&format!("/v1/sailboxes/{sailbox_id}/listeners"), &body)
            .await?;
        raise_api_error(status, &data, "")?;
        serde_json::from_value::<AddListenerWire>(data)
            .map(crate::worker::Listener::from)
            .map_err(|e| SailError::Internal {
                message: format!("failed to parse add-listener response: {e}"),
            })
    }

    /// Remove a runtime ingress listener
    /// (`DELETE /v1/sailboxes/{id}/listeners/{guest_port}`).
    ///
    /// Runs without retry so a lost response on a committed delete does not
    /// re-issue the call and surface an already-unexposed port as an error.
    pub async fn unexpose(&self, sailbox_id: &str, guest_port: u32) -> Result<(), SailError> {
        // No retry: a committed DELETE whose response is lost would, on retry,
        // hit the scheduler as an already-unexposed port (NotFound) and raise
        // even though the port was removed.
        let (status, data) = self
            .request(
                Method::Delete,
                &format!("/v1/sailboxes/{sailbox_id}/listeners/{guest_port}"),
                &[],
                None,
                NO_RETRY,
                None,
            )
            .await?;
        raise_api_error(status, &data, "")
    }

    /// List a sailbox's ingress listeners
    /// (`GET /v1/sailboxes/{id}/listeners`). Served from the control plane, so
    /// it does not resume (wake) a paused or sleeping box.
    pub async fn list_listeners(&self, sailbox_id: &str) -> Result<Vec<Listener>, SailError> {
        let (status, data) = self
            .get_request(&format!("/v1/sailboxes/{sailbox_id}/listeners"), &[])
            .await?;
        raise_api_error(status, &data, &format!("Sailbox {sailbox_id:?}"))?;
        let rows = data.get("data").cloned().unwrap_or(Value::Null);
        serde_json::from_value(rows).map_err(|e| SailError::Internal {
            message: format!("failed to parse listeners: {e}"),
        })
    }

    /// Fetch one ingress listener by guest port
    /// (`GET /v1/sailboxes/{id}/listeners/{guest_port}`); a missing port is a
    /// [`SailError::NotFound`]. Served without resuming the box.
    pub async fn get_listener(
        &self,
        sailbox_id: &str,
        guest_port: u32,
    ) -> Result<Listener, SailError> {
        let (status, data) = self
            .get_request(
                &format!("/v1/sailboxes/{sailbox_id}/listeners/{guest_port}"),
                &[],
            )
            .await?;
        raise_api_error(
            status,
            &data,
            &format!("Listener {sailbox_id:?}:{guest_port}"),
        )?;
        serde_json::from_value(data).map_err(|e| SailError::Internal {
            message: format!("failed to parse listener: {e}"),
        })
    }

    /// Ingress-identity headers for this sailbox.
    pub async fn ingress_auth_headers(
        &self,
        sailbox_id: &str,
    ) -> Result<Vec<(String, String)>, SailError> {
        let (status, data) = self
            .get_request(&format!("/v1/sailboxes/{sailbox_id}/ingress-auth"), &[])
            .await?;
        raise_api_error(status, &data, &format!("Sailbox {sailbox_id:?}"))?;
        let headers = data
            .get("headers")
            .and_then(Value::as_object)
            .ok_or_else(|| missing_field("headers"))?;
        Ok(headers
            .iter()
            .map(|(k, v)| (k.clone(), v.as_str().unwrap_or("").to_string()))
            .collect())
    }

    // --- NFS volumes (sailbox-volume API on the same host) ---

    /// Look up an NFS volume by name (`POST /v1/sailbox-volumes/get`),
    /// optionally creating it when missing.
    ///
    /// `name` must be non-empty. When `mint_if_missing` is set the volume is
    /// created if it does not yet exist.
    pub async fn get_volume(
        &self,
        name: &str,
        mint_if_missing: bool,
    ) -> Result<VolumeInfo, SailError> {
        if name.trim().is_empty() {
            return Err(SailError::InvalidArgument {
                message: "name is required".to_string(),
            });
        }
        let body = json!({
            "name": name,
            "backend": "nfs",
            "mint_if_missing": mint_if_missing,
            "create_if_missing": mint_if_missing,
        });
        let (status, data) = self.post("/v1/sailbox-volumes/get", &body).await?;
        raise_api_error(status, &data, "")?;
        volume_from(data)
    }

    /// List the org's NFS volumes (`GET /v1/sailbox-volumes`).
    ///
    /// `max_objects`, when given, caps the number returned and must be
    /// non-negative.
    pub async fn list_volumes(
        &self,
        max_objects: Option<i64>,
    ) -> Result<Vec<VolumeInfo>, SailError> {
        let mut query: Vec<(String, String)> = Vec::new();
        if let Some(max) = max_objects {
            if max < 0 {
                return Err(SailError::InvalidArgument {
                    message: "max_objects cannot be negative".to_string(),
                });
            }
            query.push(("max_objects".to_string(), max.to_string()));
        }
        let (status, data) = self.get_request("/v1/sailbox-volumes", &query).await?;
        raise_api_error(status, &data, "")?;
        data.get("volumes").and_then(Value::as_array).map_or_else(
            || Ok(Vec::new()),
            |rows| rows.iter().cloned().map(volume_from).collect(),
        )
    }

    /// Delete a volume by id; `Ok(None)` on a 204 (no body), the deleted handle
    /// otherwise.
    pub async fn delete_volume(
        &self,
        volume_id: &str,
        allow_missing: bool,
    ) -> Result<Option<VolumeInfo>, SailError> {
        if volume_id.trim().is_empty() {
            return Err(SailError::InvalidArgument {
                message: "volume_id is required".to_string(),
            });
        }
        let path = format!("/v1/sailbox-volumes/{}", volume_id.trim());
        let query: Vec<(String, String)> = if allow_missing {
            vec![("allow_missing".to_string(), "true".to_string())]
        } else {
            Vec::new()
        };
        let policy = if allow_missing {
            DEFAULT_RETRY_POLICY
        } else {
            NO_RETRY
        };
        let (status, data) = self
            .request(Method::Delete, &path, &query, None, policy, None)
            .await?;
        if status == 204 {
            return Ok(None);
        }
        raise_api_error(status, &data, "")?;
        Ok(Some(volume_from(data)?))
    }

    /// Fetch the caller org's SSH certificate-authority public key, creating the
    /// CA on first use. A box trusts this key, so installing it is what lets the
    /// org's signed user certificates authenticate.
    pub async fn org_ssh_ca_public_key(&self) -> Result<String, SailError> {
        let (status, data) = self.get_request("/v1/ssh/ca", &[]).await?;
        raise_api_error(status, &data, "ssh ca")?;
        str_field(&data, "public_key")
    }

    /// Exchange a local SSH public key for a short-lived certificate the org CA
    /// signs (principal `root`). Returns the OpenSSH certificate and its key id
    /// (`org=<id>;fp=...;iat=...`), which identifies the signing org.
    ///
    /// `timeout` bounds a single attempt with no retry (used by the auto-refresh
    /// hook, which runs inside the ssh connect path and must fail open quickly);
    /// `None` uses the default retrying transport.
    pub async fn issue_user_cert(
        &self,
        public_key: &str,
        timeout: Option<f64>,
    ) -> Result<IssuedUserCert, SailError> {
        let body = json!({ "public_key": public_key });
        let bytes = serde_json::to_vec(&body).map_err(|e| SailError::Internal {
            message: format!("failed to serialize request body: {e}"),
        })?;
        let policy = if timeout.is_some() {
            NO_RETRY
        } else {
            DEFAULT_RETRY_POLICY
        };
        let (status, data) = self
            .request(
                Method::Post,
                "/v1/ssh/certificate",
                &[],
                Some(bytes),
                policy,
                timeout,
            )
            .await?;
        raise_api_error(status, &data, "ssh certificate")?;
        Ok(IssuedUserCert {
            certificate: str_field(&data, "certificate")?,
            key_id: data
                .get("key_id")
                .and_then(Value::as_str)
                .unwrap_or_default()
                .to_string(),
        })
    }

    // --- transport helpers ---

    async fn post(&self, path: &str, body: &Value) -> Result<(u16, Value), SailError> {
        let bytes = serde_json::to_vec(body).map_err(|e| SailError::Internal {
            message: format!("failed to serialize request body: {e}"),
        })?;
        self.request(
            Method::Post,
            path,
            &[],
            Some(bytes),
            DEFAULT_RETRY_POLICY,
            None,
        )
        .await
    }

    async fn get_request(
        &self,
        path: &str,
        query: &[(String, String)],
    ) -> Result<(u16, Value), SailError> {
        self.request(Method::Get, path, query, None, DEFAULT_RETRY_POLICY, None)
            .await
    }

    async fn request(
        &self,
        method: Method,
        path: &str,
        query: &[(String, String)],
        body: Option<Vec<u8>>,
        policy: RetryPolicy,
        timeout: Option<f64>,
    ) -> Result<(u16, Value), SailError> {
        let spec = RequestSpec {
            method,
            path: path.to_string(),
            query: query.to_vec(),
            body,
            extra_headers: Vec::new(),
            timeout,
            policy,
            idempotency_key: IdempotencyKey::Auto,
        };
        self.http.request(&spec).await
    }
}

fn handle_from(data: &Value, fallback_name: &str) -> SailboxHandle {
    SailboxHandle {
        sailbox_id: data
            .get("sailbox_id")
            .and_then(Value::as_str)
            .unwrap_or("")
            .to_string(),
        name: data
            .get("name")
            .and_then(Value::as_str)
            .unwrap_or(fallback_name)
            .to_string(),
        status: status_of(data),
        worker_address: data
            .get("worker_address")
            .and_then(Value::as_str)
            .unwrap_or("")
            .to_string(),
        exec_endpoint: data
            .get("exec_proxy_endpoint")
            .and_then(Value::as_str)
            .unwrap_or("")
            .to_string(),
    }
}

fn info_from(data: Value) -> Result<SailboxInfo, SailError> {
    serde_json::from_value::<SailboxInfo>(data).map_err(|e| SailError::Internal {
        message: format!("failed to parse sailbox: {e}"),
    })
}

fn volume_from(data: Value) -> Result<VolumeInfo, SailError> {
    serde_json::from_value::<VolumeInfo>(data).map_err(|e| SailError::Internal {
        message: format!("failed to parse volume: {e}"),
    })
}

fn int_field(data: &Value, key: &str) -> Result<i64, SailError> {
    data.get(key)
        .and_then(Value::as_i64)
        .ok_or_else(|| missing_field(key))
}

fn str_field(data: &Value, key: &str) -> Result<String, SailError> {
    data.get(key)
        .and_then(Value::as_str)
        .filter(|s| !s.is_empty())
        .map(str::to_string)
        .ok_or_else(|| missing_field(key))
}

fn missing_field(key: &str) -> SailError {
    SailError::Internal {
        message: format!("API response missing field {key:?}"),
    }
}

/// A 404 body the API tags as a real missing resource (vs a route-level 404).
fn is_resource_not_found(data: &Value) -> bool {
    data.get("error")
        .and_then(|e| e.get("type"))
        .and_then(Value::as_str)
        == Some("not_found_error")
}

/// Map a non-2xx lifecycle/monitoring response onto the canonical taxonomy:
/// resource-miss 404 → NotFound, 401/403 → PermissionDenied, 400 →
/// InvalidArgument, any other → Api (a transient/unexpected API failure).
fn raise_api_error(status: u16, data: &Value, context: &str) -> Result<(), SailError> {
    if status < 300 {
        return Ok(());
    }
    let mut message = crate::http::api_error_message(data, "request failed");
    if !context.is_empty() {
        message = format!("{context}: {message}");
    }
    if status == 404 && is_resource_not_found(data) {
        return Err(SailError::NotFound { message });
    }
    if status == 401 || status == 403 {
        return Err(SailError::PermissionDenied { message });
    }
    if status == 400 {
        return Err(SailError::InvalidArgument { message });
    }
    Err(SailError::Api {
        status,
        message,
        body: data.clone(),
    })
}

/// create's ladder: every non-2xx except auth is a Creation failure.
fn raise_for_create_status(status: u16, data: &Value) -> Result<(), SailError> {
    if status < 300 {
        return Ok(());
    }
    let message = crate::http::api_error_message(data, "request failed");
    if status == 401 || status == 403 {
        return Err(SailError::PermissionDenied { message });
    }
    Err(SailError::Creation {
        message,
        status,
        body: data.clone(),
    })
}

/// The echoed lifecycle `status`, parsed into the typed enum.
fn status_of(data: &Value) -> SailboxStatus {
    SailboxStatus::from(data.get("status").and_then(Value::as_str).unwrap_or(""))
}

/// Raise an Api error when the echoed status is not the expected one.
fn require_status(data: &Value, expected: SailboxStatus, status: u16) -> Result<(), SailError> {
    let got = status_of(data);
    if got == expected {
        return Ok(());
    }
    Err(SailError::Api {
        status,
        message: format!("lifecycle call returned unexpected status: {got}"),
        body: data.clone(),
    })
}

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

    fn port(guest_port: u32, protocol: IngressProtocol, allowlist: &[&str]) -> IngressPort {
        IngressPort {
            guest_port,
            protocol,
            allowlist: allowlist.iter().map(ToString::to_string).collect(),
        }
    }

    fn mount(volume_id: &str, mount_path: &str) -> VolumeMount {
        VolumeMount {
            volume_id: volume_id.to_string(),
            mount_path: mount_path.to_string(),
        }
    }

    #[test]
    fn ingress_ports_accepts_valid_and_rejects_the_rules() {
        assert!(validate_ingress_ports(&[
            port(8080, IngressProtocol::Http, &[]),
            port(22, IngressProtocol::Tcp, &[]),
            port(5432, IngressProtocol::Tcp, &["10.0.0.0/8"]),
        ])
        .is_ok());
        // http/22 reserved for ssh.
        assert!(validate_ingress_ports(&[port(22, IngressProtocol::Http, &[])]).is_err());
        // agent infra port.
        assert!(validate_ingress_ports(&[port(10000, IngressProtocol::Tcp, &[])]).is_err());
        // out of range.
        assert!(validate_ingress_ports(&[port(70000, IngressProtocol::Http, &[])]).is_err());
        // duplicate guest port across protocols.
        assert!(validate_ingress_ports(&[
            port(8080, IngressProtocol::Http, &[]),
            port(8080, IngressProtocol::Tcp, &["0.0.0.0/0"]),
        ])
        .is_err());
        // app-name allowlist on a tcp listener.
        assert!(validate_ingress_ports(&[port(9000, IngressProtocol::Tcp, &["my-app"])]).is_err());
        // an app-name allowlist on http is fine.
        assert!(validate_ingress_ports(&[port(9000, IngressProtocol::Http, &["my-app"])]).is_ok());
        // unauthenticated service port needs an allowlist.
        assert!(validate_ingress_ports(&[port(5432, IngressProtocol::Tcp, &[])]).is_err());
        // invalid CIDR.
        assert!(
            validate_ingress_ports(&[port(9000, IngressProtocol::Http, &["1.2.3.4/33"])]).is_err()
        );
    }

    #[test]
    fn volume_mounts_reject_root_reserved_and_overlaps() {
        assert!(validate_volume_mounts(&[mount("vol_1", "/mnt/data")]).is_ok());
        assert!(validate_volume_mounts(&[mount("", "/mnt/data")]).is_err());
        assert!(validate_volume_mounts(&[mount("vol_1", "relative")]).is_err());
        assert!(validate_volume_mounts(&[mount("vol_1", "/")]).is_err());
        assert!(validate_volume_mounts(&[mount("vol_1", "/proc/x")]).is_err());
        assert!(
            validate_volume_mounts(&[mount("vol_1", "/mnt"), mount("vol_2", "/mnt/cache"),])
                .is_err()
        );
        // normpath resolves .. before the overlap/reserved checks.
        assert!(validate_volume_mounts(&[mount("vol_1", "/mnt/../proc")]).is_err());
    }

    #[test]
    fn info_reads_resource_fields_and_defaults_optionals() {
        let info = info_from(json!({
            "sailbox_id": "sb-1", "app_id": "app-1", "app_name": "a", "name": "n",
            "image_id": "img-1",
            "status": "running", "memory_mib": 2048, "vcpu_count": 4,
            "state_disk_size_gib": 10,
            "cpu_requested_vcpu": 2, "cpu_used_vcpu": 1.5,
            "memory_requested_bytes": 1024, "memory_used_bytes": 512,
            "disk_requested_bytes": 4096, "disk_used_bytes": 2048,
            "architecture": "amd64", "checkpoint_generation": 7,
            "created_at": "2026-01-01T00:00:00Z", "updated_at": "2026-01-02T00:00:00Z"
        }))
        .unwrap();
        // Resource fields are read straight from the wire (the backend always sends them).
        assert_eq!(info.cpu_requested_vcpu, 2);
        assert_eq!(info.memory_used_bytes, 512);
        assert_eq!(info.checkpoint_generation, 7);
        // Genuinely-optional fields default to None when absent.
        assert_eq!(info.guest_schema_version, None);
        assert_eq!(info.error_message, None);
        assert_eq!(info.started_at, None);
    }

    #[test]
    fn api_error_ladder_maps_statuses() {
        let not_found = json!({"error": {"type": "not_found_error", "message": "gone"}});
        assert!(matches!(
            raise_api_error(404, &not_found, ""),
            Err(SailError::NotFound { .. })
        ));
        // A route-level 404 (no not_found_error type) is transient, not a miss.
        let route_404 = json!({"error": {"message": "no route"}});
        assert!(matches!(
            raise_api_error(404, &route_404, ""),
            Err(SailError::Api { .. })
        ));
        let auth = json!({"error": {"message": "nope"}});
        assert!(matches!(
            raise_api_error(403, &auth, ""),
            Err(SailError::PermissionDenied { .. })
        ));
        assert!(matches!(
            raise_api_error(400, &auth, ""),
            Err(SailError::InvalidArgument { .. })
        ));
        assert!(matches!(
            raise_api_error(503, &auth, ""),
            Err(SailError::Api { status: 503, .. })
        ));
        assert!(raise_api_error(200, &json!({}), "").is_ok());
    }

    #[test]
    fn create_ladder_maps_non_auth_to_creation() {
        let body = json!({"error": {"message": "no capacity"}});
        assert!(matches!(
            raise_for_create_status(503, &body),
            Err(SailError::Creation { .. })
        ));
        assert!(matches!(
            raise_for_create_status(401, &body),
            Err(SailError::PermissionDenied { .. })
        ));
    }

    // The Rust half of the cross-language image contract: ImageSpec's serde
    // attributes must produce canonical proto-JSON (camelCase fields, enum value
    // names, flattened oneofs). The Go side proves protojson decodes this exact
    // shape (TestCreateSailboxHandlerAcceptsCanonicalProtoJSON). Keep the two in sync.
    #[test]
    fn image_spec_serializes_to_canonical_proto_json() {
        use crate::image::{
            BaseImage, ImageArchitecture, ImageBuildStep, ImageSpec, PackageInstall, RunCommand,
        };
        let spec = ImageSpec {
            base: Some(BaseImage::Debian),
            architecture: ImageArchitecture::Arm64,
            python_version: "3.12".to_string(),
            build_steps: vec![
                ImageBuildStep::AptInstall(PackageInstall {
                    packages: vec!["git".to_string(), "curl".to_string()],
                }),
                ImageBuildStep::RunCommand(RunCommand {
                    command: "echo hi".to_string(),
                }),
            ],
            ..Default::default()
        };
        let json = serde_json::to_value(&spec).unwrap();
        assert_eq!(json["base"], json!("BASE_IMAGE_DEBIAN"));
        assert_eq!(json["architecture"], json!("IMAGE_ARCHITECTURE_ARM64"));
        assert_eq!(json["pythonVersion"], json!("3.12"));
        assert_eq!(
            json["buildSteps"][0]["aptInstall"]["packages"][0],
            json!("git")
        );
        assert_eq!(
            json["buildSteps"][1]["runCommand"]["command"],
            json!("echo hi")
        );

        let devbox = ImageSpec {
            base: Some(BaseImage::Devbox),
            architecture: ImageArchitecture::Arm64,
            ..Default::default()
        };
        let devbox_json = serde_json::to_value(&devbox).unwrap();
        assert_eq!(devbox_json["base"], json!("BASE_IMAGE_DEVBOX"));
    }
}