sail-rs 0.6.3

//! 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 percent_encoding::{utf8_percent_encode, AsciiSet, CONTROLS};
use serde_json::{json, Value};
use time::format_description::well_known::Rfc3339;
use time::OffsetDateTime;

use crate::apierror::{is_resource_not_found, raise_api_error};
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, AutoSleep, CreateSailboxRequest, CustomDomainInfo, IngressPort,
    IngressProtocol, IssuedUserCert, ListSailboxesQuery, SailboxCheckpoint, SailboxHandle,
    SailboxInfo, SailboxListOrder, SailboxMetricsQuery, SailboxMetricsResponse, SailboxPage,
    SailboxSpendQuery, SailboxSpendResponse, SailboxStatus, VolumeInfo, VolumeMount, WhoAmI,
};
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"];

// RFC 3986 path-segment delimiters plus controls. Domain punctuation such as
// dots and hyphens stays readable while Unicode and route-breaking bytes are
// encoded before a hostname is placed in a DELETE path.
const PATH_SEGMENT_ENCODE_SET: &AsciiSet = &CONTROLS
    .add(b' ')
    .add(b'"')
    .add(b'#')
    .add(b'%')
    .add(b'/')
    .add(b'?')
    .add(b'\\')
    .add(b'{')
    .add(b'}');

/// Longest accepted Sailbox name. A name is a display label, the fallback
/// lookup key when only the name is known, and the host in `ssh <name>.sail`.
const MAX_SAILBOX_NAME_LEN: usize = 128;

/// The longest wait a caller may ask for before an automatic sleep. Past an
/// hour the wait stops being a delay and becomes a second way to spell
/// [`AutoSleep::Never`], which says it more clearly. The backend enforces the
/// same bound; checking here means every SDK fails the same way without a round
/// trip.
pub(crate) const MAX_AUTO_SLEEP_WAIT: Duration = Duration::from_hours(1);

/// Reject a wait the backend would refuse.
pub fn validate_auto_sleep(auto_sleep: AutoSleep) -> Result<(), SailError> {
    if let AutoSleep::NotBefore(wait) = auto_sleep {
        if wait > MAX_AUTO_SLEEP_WAIT {
            return Err(SailError::InvalidArgument {
                message: format!(
                    "auto-sleep wait must be at most {} seconds; turn automatic sleep off instead",
                    MAX_AUTO_SLEEP_WAIT.as_secs()
                ),
            });
        }
    }
    Ok(())
}

/// Validate a caller-supplied Sailbox name: length-bounded and free of
/// control characters. Deliberately permissive beyond that (existing names
/// with spaces or punctuation stay valid).
pub fn validate_sailbox_name(name: &str) -> Result<(), SailError> {
    if name.chars().count() > MAX_SAILBOX_NAME_LEN {
        return Err(SailError::InvalidArgument {
            message: format!("name must be at most {MAX_SAILBOX_NAME_LEN} characters"),
        });
    }
    if name.chars().any(char::is_control) {
        return Err(SailError::InvalidArgument {
            message: "name must not contain control characters".to_string(),
        });
    }
    Ok(())
}

/// Largest whole-seconds value the wire carries. `ttl_seconds` decodes as a
/// uint32 server-side.
const MAX_WIRE_SECONDS: i64 = u32::MAX as i64;

/// Largest `timeout_seconds` start-from-checkpoint accepts. The server stores
/// it in a column holding a 32-bit signed integer.
const MAX_STORED_LIFETIME_SECONDS: i64 = i32::MAX as i64;

/// Bound a whole-seconds value before it goes on the wire, so a value the
/// server would turn away fails locally.
fn wire_seconds(name: &str, value: i64, max: i64) -> Result<i64, SailError> {
    if value <= 0 || value > max {
        return Err(SailError::InvalidArgument {
            message: format!("{name} must be between 1 and {max}"),
        });
    }
    Ok(value)
}

/// Parse an allowlist entry as an address range. `ipnet` takes zero-padded
/// octets and prefix lengths the server refuses, such as "01.2.3.4/24" and
/// "0.0.0.0/00". So the address half goes through the strict parser, a prefix
/// length past one character has to open with 1-9 the way the server demands,
/// and only the masking is left to `ipnet`.
fn parse_range(value: &str) -> Option<ipnet::IpNet> {
    let (addr, prefix_len) = value.split_once('/')?;
    if prefix_len.len() > 1 && !matches!(prefix_len.as_bytes()[0], b'1'..=b'9') {
        return None;
    }
    addr.parse::<std::net::IpAddr>().ok()?;
    value.parse::<ipnet::IpNet>().ok()
}

/// 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 {
    parse_range(value).is_some() || value.parse::<std::net::IpAddr>().is_ok()
}

/// Whether an allowlist entry carries an IPv6 zone, as in "fe80::1%eth0". A
/// zone names an interface on the host holding the address, which says nothing
/// about a remote source, and it runs to the end of the entry, so
/// "fe80::1%eth0/64" reads as a zone of "eth0/64" with no prefix length left.
/// An entry whose zone is empty, as in "fe80::1%", is not zoned: it reads as
/// neither an address nor a range, so it stays an app-name entry, which is how
/// the server stores it.
fn is_zoned(value: &str) -> bool {
    match value.split_once('%') {
        Some((addr, zone)) => !zone.is_empty() && addr.parse::<std::net::Ipv6Addr>().is_ok(),
        None => false,
    }
}

/// 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),
    ("l", 16, 256, 64, 1024),
];

/// 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_limit_gib: Option<u32>,
    disk_limit_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_limit_gib) = memory_limit_gib {
        if !(mem_min..=mem_max).contains(&memory_limit_gib) {
            return Err(SailError::InvalidArgument {
                message: format!(
                    "memory_limit_gib for size {label} must be between {mem_min} and {mem_max}"
                ),
            });
        }
    }
    if let Some(disk_limit_gib) = disk_limit_gib {
        if !(disk_min..=disk_max).contains(&disk_limit_gib) {
            return Err(SailError::InvalidArgument {
                message: format!(
                    "disk_limit_gib for size {label} must be between {disk_min} and {disk_max}"
                ),
            });
        }
    }
    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
/// zoned address and a slash-containing entry that is not a valid CIDR, and
/// dedupe. App-name entries pass through verbatim.
///
/// Address syntax wins over app names here and on the server, so an app named
/// like an address or a prefix cannot be named in an allowlist.
pub(crate) 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;
        }
        if is_zoned(value) {
            return Err(SailError::InvalidArgument {
                message: format!("allowlist entry {value:?} must not carry an IPv6 zone"),
            });
        }
        let normalized = if let Some(net) = parse_range(value) {
            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 address range"),
            });
        } else {
            value.to_string()
        };
        if seen.insert(normalized.clone()) {
            out.push(normalized);
        }
    }
    Ok(out)
}

/// Validate ingress ports against the server's per-protocol rules (reserved
/// ports, 1..65535 range, unique guest port, allowlist shape). The server
/// enforces the same rules, so an invalid list is rejected here before the
/// call is sent. 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;
            if is_zoned(value) {
                return invalid(format!(
                    "allowlist entry {value:?} must not carry an IPv6 zone"
                ));
            }
            let is_cidr = is_cidr_or_ip(value);
            // A '/' is read as a CIDR, so a malformed prefix is a typo rather
            // than an app name.
            if value.contains('/') && !is_cidr {
                return invalid(format!(
                    "allowlist entry {value:?} is not a valid address range"
                ));
            }
            // 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; a tcp allowlist takes an address or a range"
                ));
            }
        }
        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 by the Sailbox runtime 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 addresses or ranges to restrict sources (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 a Sailbox runtime upgrade.
#[derive(Debug, Clone, serde::Serialize)]
#[non_exhaustive]
pub struct UpgradeResult {
    /// True when no upgrade is left to apply, either because the Sailbox took
    /// one just now or because it was already current. False when the upgrade
    /// is recorded and takes effect the next time the Sailbox wakes.
    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 Sailbox 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 Sailbox behind under the same name. `None`
    /// leaves each attempt unbounded. If the budget is exhausted the Sailbox
    /// 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_sailbox_name(&req.name)?;
        validate_auto_sleep(req.auto_sleep)?;
        validate_size_limits(
            req.size.map(|s| s.as_str()),
            req.memory_limit_gib,
            req.disk_limit_gib,
        )?;
        validate_ingress_ports(&req.ingress_ports)?;
        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_limit_gib) = req.memory_limit_gib {
            body["memory_limit_gib"] = json!(memory_limit_gib);
        }
        if let Some(disk_limit_gib) = req.disk_limit_gib {
            body["state_disk_limit_gib"] = json!(disk_limit_gib);
        }
        if req.private {
            body["visibility"] = json!("private");
        }
        if req.auto_sleep != AutoSleep::Automatic {
            body["auto_sleep"] = req.auto_sleep.to_json();
        }
        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 identity behind the API key (`GET /v1/whoami`).
    pub async fn whoami(&self) -> Result<WhoAmI, SailError> {
        let (status, data) = self.get_request("/v1/whoami", &[]).await?;
        raise_api_error(status, &data, "")?;
        serde_json::from_value(data).map_err(|e| SailError::Internal {
            message: format!("failed to parse whoami: {e}"),
        })
    }

    /// 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, and search 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 query.order != SailboxListOrder::NewestActive {
            params.push(("order".to_string(), query.order.as_str().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(id) = &query.credential_policy_id {
            params.push(("credential_injection_policy_id".to_string(), id.clone()));
        }
        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,
        })
    }

    /// Estimate Sailbox spend over a time window (`GET /v1/sailboxes/spend`).
    pub async fn spend(
        &self,
        query: &SailboxSpendQuery,
    ) -> Result<SailboxSpendResponse, SailError> {
        let mut params: Vec<(String, String)> = Vec::new();
        if let Some(app_id) = &query.app_id {
            params.push(("app_id".to_string(), app_id.clone()));
        }
        if let Some(sailbox_id) = &query.sailbox_id {
            params.push(("sailbox_id".to_string(), sailbox_id.clone()));
        }
        if let Some(from) = query.from {
            params.push((
                "from".to_string(),
                from.format(&Rfc3339).map_err(|err| SailError::Internal {
                    message: format!("failed to format spend start timestamp: {err}"),
                })?,
            ));
        }
        if let Some(to) = query.to {
            params.push((
                "to".to_string(),
                to.format(&Rfc3339).map_err(|err| SailError::Internal {
                    message: format!("failed to format spend end timestamp: {err}"),
                })?,
            ));
        }
        let (status, data) = self.get_request("/v1/sailboxes/spend", &params).await?;
        raise_api_error(status, &data, "Sailbox spend")?;
        serde_json::from_value(data).map_err(|err| SailError::Internal {
            message: format!("failed to parse sailbox spend response: {err}"),
        })
    }

    /// Fetch a Sailbox's resource-usage time series (`GET /v1/sailboxes/{id}/metrics`).
    pub async fn metrics(
        &self,
        sailbox_id: &str,
        query: &SailboxMetricsQuery,
    ) -> Result<SailboxMetricsResponse, SailError> {
        let params = vec![("range".to_string(), query.range.clone())];
        let (status, data) = self
            .get_request(&format!("/v1/sailboxes/{sailbox_id}/metrics"), &params)
            .await?;
        raise_api_error(status, &data, "Sailbox metrics")?;
        serde_json::from_value(data).map_err(|err| SailError::Internal {
            message: format!("failed to parse sailbox metrics response: {err}"),
        })
    }

    /// 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, optionally scheduling a wall-clock wake first.
    ///
    /// When `wake_at` is given, the wake is recorded before the sleep starts
    /// and survives it: if the Sailbox is sleeping when the moment arrives,
    /// Sail restores it. A request earlier than the current scheduled wake
    /// replaces it; a later request leaves the sooner wake in place. The
    /// returned time is the effective (sooner) wake. A wake can fire a
    /// little after the time you set, so treat it as approximate. Calling
    /// sleep on a Sailbox that is already sleeping succeeds and just updates
    /// the scheduled wake.
    ///
    /// Requires the echoed status to be `sleeping`.
    pub async fn sleep(
        &self,
        sailbox_id: &str,
        wake_at: Option<OffsetDateTime>,
    ) -> Result<Option<OffsetDateTime>, SailError> {
        // Record the wake before the sleep starts: a wake recorded on a
        // running box survives its sleep, while one scheduled mid-capture
        // can be rejected until the capture settles.
        let effective = match wake_at {
            Some(when) => Some(self.wake_at(sailbox_id, when).await?),
            None => None,
        };
        self.stop(sailbox_id, "sleep", SailboxStatus::Sleeping)
            .await?;
        Ok(effective)
    }

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

    // Schedule a wall-clock wake (`POST /v1/sailboxes/{id}/wake_at`),
    // returning the effective (sooner-of-the-two) wake time. Reached only
    // through sleep's wake_at option: recording the wake before the sleep is
    // what makes "sleep until" race-free, and paused Sailboxes reject
    // scheduled wakes (explicit resumes only).
    async fn wake_at(
        &self,
        sailbox_id: &str,
        when: OffsetDateTime,
    ) -> Result<OffsetDateTime, SailError> {
        let formatted = when.format(&Rfc3339).map_err(|err| SailError::Internal {
            message: format!("failed to format wake time: {err}"),
        })?;
        let (status, data) = self
            .post(
                &format!("/v1/sailboxes/{sailbox_id}/wake_at"),
                &json!({"wake_at": formatted}),
            )
            .await?;
        raise_api_error(status, &data, "")?;
        data.get("wake_at")
            .and_then(Value::as_str)
            .and_then(|s| OffsetDateTime::parse(s, &Rfc3339).ok())
            .ok_or_else(|| SailError::Api {
                status,
                message: "wake response has a missing or malformed effective wake time".to_string(),
                body: data.clone(),
            })
    }

    /// Replace when Sail may sleep this Sailbox on its own
    /// (`POST /v1/sailboxes/{id}/auto_sleep`).
    ///
    /// Each call replaces the whole setting: switching to
    /// [`Never`](AutoSleep::Never) clears any minimum wait set earlier, and
    /// switching back does not restore it.
    pub async fn set_auto_sleep(
        &self,
        sailbox_id: &str,
        auto_sleep: AutoSleep,
    ) -> Result<(), SailError> {
        validate_auto_sleep(auto_sleep)?;
        let (status, data) = self
            .post(
                &format!("/v1/sailboxes/{sailbox_id}/auto_sleep"),
                &auto_sleep.to_json(),
            )
            .await?;
        raise_api_error(status, &data, "")?;
        Ok(())
    }

    /// 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 fit a uint32, 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 {
            body["ttl_seconds"] = json!(wire_seconds("ttl_seconds", ttl, MAX_WIRE_SECONDS)?);
        }
        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 and at most 2147483647; both are validated as
    /// [`SailError::InvalidArgument`] before the call. Requires a `running`
    /// status and returns the new Sailbox handle.
    // CR-soon charley: `timeout_seconds` reaches the backend, which stores it on
    // the new row and never reads it back, and the call does not pass a
    // client-side deadline. Drop the parameter from the API and the ergonomic
    // surfaces above it in a breaking release.
    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(),
            });
        }
        if let Some(name) = name {
            validate_sailbox_name(name)?;
        }
        let mut body = json!({ "checkpoint_id": checkpoint_id });
        if let Some(name) = name {
            body["name"] = json!(name);
        }
        if let Some(timeout) = timeout_seconds {
            body["timeout_seconds"] = json!(wire_seconds(
                "timeout",
                timeout,
                MAX_STORED_LIFETIME_SECONDS
            )?);
        }
        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 runtime (`POST /v1/sailboxes/{id}/upgrade`).
    ///
    /// The returned [`UpgradeResult`] reports whether anything is left to
    /// apply, or whether the upgrade is recorded for 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 unknown 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> {
        // The shared validator's message names ingress_ports (the create-time
        // parameter); expose takes one port, so range-check it under its own
        // parameter name first.
        if guest_port == 0 || guest_port > 65535 {
            return Err(SailError::InvalidArgument {
                message: format!("guest_port must be between 1 and 65535, got {guest_port}"),
            });
        }
        // The same rules create-time ingress ports get, so an invalid port or
        // allowlist fails fast as an invalid argument instead of surfacing as
        // a transport-level rejection.
        validate_ingress_ports(&[IngressPort {
            guest_port,
            protocol,
            allowlist: allowlist.to_vec(),
        }])?;
        // 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}"),
                &[],
                /* body */ None,
                NO_RETRY,
                /* timeout */ 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 Sailbox.
    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 Sailbox.
    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}"),
        })
    }

    /// Fetch the current organization's custom-domain DNS targets.
    #[doc(hidden)]
    pub async fn custom_domain_dns_targets(&self) -> Result<(String, Option<String>), SailError> {
        let (status, data) = self.get_request("/v1/custom-domains", &[]).await?;
        raise_api_error(status, &data, "custom domain DNS configuration")?;
        let cname_target = str_field(&data, "cname_target")?;
        let acme_challenge_target = data
            .get("acme_challenge_target")
            .and_then(Value::as_str)
            .filter(|target| !target.is_empty())
            .map(str::to_owned);
        Ok((cname_target, acme_challenge_target))
    }

    /// 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<CustomDomainInfo, SailError> {
        if guest_port == 0 || guest_port > 65535 {
            return Err(SailError::InvalidArgument {
                message: format!("guest_port must be between 1 and 65535, got {guest_port}"),
            });
        }
        let body = json!({"domain": domain, "guest_port": guest_port});
        let (status, data) = self
            .post(&format!("/v1/sailboxes/{sailbox_id}/domains"), &body)
            .await?;
        raise_api_error(status, &data, "custom domain")?;
        custom_domain_from(data)
    }

    /// List the custom domains attached to a Sailbox.
    #[doc(hidden)]
    pub async fn list_custom_domains(
        &self,
        sailbox_id: &str,
    ) -> Result<Vec<CustomDomainInfo>, SailError> {
        let (status, data) = self
            .get_request(&format!("/v1/sailboxes/{sailbox_id}/domains"), &[])
            .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 custom domains: {e}"),
        })
    }

    /// Detach a custom domain from a Sailbox.
    #[doc(hidden)]
    pub async fn detach_custom_domain(
        &self,
        sailbox_id: &str,
        domain: &str,
    ) -> Result<(), SailError> {
        let domain = utf8_percent_encode(domain, PATH_SEGMENT_ENCODE_SET);
        let (status, data) = self
            .request(
                Method::Delete,
                &format!("/v1/sailboxes/{sailbox_id}/domains/{domain}"),
                &[],
                /* body */ None,
                NO_RETRY,
                /* timeout */ None,
            )
            .await?;
        raise_api_error(status, &data, "custom domain")
    }

    /// 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, 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> {
        let name = name.trim();
        if name.is_empty() {
            return Err(SailError::InvalidArgument {
                message: "name is required".to_string(),
            });
        }
        if mint_if_missing {
            // Names are unique within an organization, so a create under a name
            // that is taken answers with the volume already under it.
            let body = json!({ "name": name });
            let (status, data) = self.post("/v1/sailbox-volumes", &body).await?;
            raise_api_error(status, &data, "")?;
            return volume_from(data);
        }
        let query = vec![("name".to_string(), name.to_string())];
        let (status, data) = self.get_request("/v1/sailbox-volumes", &query).await?;
        raise_api_error(status, &data, "")?;
        // A name that matches nothing comes back as an empty page.
        data.get("data")
            .and_then(Value::as_array)
            .and_then(|rows| rows.first())
            .cloned()
            .map_or_else(
                || {
                    Err(SailError::NotFound {
                        message: "volume not found".to_string(),
                    })
                },
                volume_from,
            )
    }

    /// 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(("limit".to_string(), max.to_string()));
        }
        let (status, data) = self.get_request("/v1/sailbox-volumes", &query).await?;
        raise_api_error(status, &data, "")?;
        data.get("data").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,
                /* body */ None,
                policy,
                /* timeout */ 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 Sailbox 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,
            /* timeout */ None,
        )
        .await
    }

    async fn get_request(
        &self,
        path: &str,
        query: &[(String, String)],
    ) -> Result<(u16, Value), SailError> {
        self.request(
            Method::Get,
            path,
            query,
            /* body */ None,
            DEFAULT_RETRY_POLICY,
            /* timeout */ 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> {
    let info = serde_json::from_value::<SailboxInfo>(data).map_err(|e| SailError::Internal {
        message: format!("failed to parse sailbox: {e}"),
    })?;
    if let Some(deprecation) = &info.deprecation {
        crate::notice::notify(
            crate::notice::NoticeKind::GuestDeprecation,
            &deprecation.message,
        );
    }
    Ok(info)
}

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 custom_domain_from(data: Value) -> Result<CustomDomainInfo, SailError> {
    serde_json::from_value(data).map_err(|e| SailError::Internal {
        message: format!("failed to parse custom domain: {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:?}"),
    }
}

/// 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::apierror::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()
        );
        // a zoned address, with and without a prefix length behind the zone.
        assert!(
            validate_ingress_ports(&[port(9000, IngressProtocol::Http, &["fe80::1%eth0"])])
                .is_err()
        );
        // The entry behind the zone also holds a '/', so pin the reason it is
        // refused.
        let err =
            validate_ingress_ports(&[port(9000, IngressProtocol::Http, &["fe80::1%eth0/64"])])
                .unwrap_err();
        assert!(err.to_string().contains("IPv6 zone"), "{err}");
        // a '%' outside an address is an app name like any other.
        assert!(validate_ingress_ports(&[port(9000, IngressProtocol::Http, &["my%app"])]).is_ok());
    }

    #[test]
    fn normalize_allowlist_canonicalizes_and_rejects_zones() {
        let entries: Vec<String> = ["  10.0.0.5/8 ", "1.2.3.4", "10.1.2.3/8", "my-app"]
            .iter()
            .map(ToString::to_string)
            .collect();
        assert_eq!(
            normalize_allowlist(&entries).unwrap(),
            vec!["10.0.0.0/8", "1.2.3.4/32", "my-app"]
        );
        assert!(normalize_allowlist(&["fe80::1%eth0".to_string()]).is_err());
        let err = normalize_allowlist(&["fe80::1%eth0/64".to_string()]).unwrap_err();
        assert!(err.to_string().contains("IPv6 zone"), "{err}");
        // A '%' outside an address, and an empty zone, are app names like any
        // other, which is how the server stores them.
        assert_eq!(
            normalize_allowlist(&["my%app".to_string()]).unwrap(),
            vec!["my%app"]
        );
        assert_eq!(
            normalize_allowlist(&["fe80::1%".to_string()]).unwrap(),
            vec!["fe80::1%"]
        );
        // The server refuses a zero-padded or signed octet or prefix length
        // outright, so these entries have to fail rather than normalize.
        for entry in [
            "01.2.3.4/24",
            "1.2.3.04/24",
            "010.1.2.3/24",
            "0.0.0.0/00",
            "203.0.113.1/+24",
            "203.0.113.1/-24",
        ] {
            let err = normalize_allowlist(&[entry.to_string()]).unwrap_err();
            assert!(err.to_string().contains("address range"), "{entry}: {err}");
            assert!(
                validate_ingress_ports(&[port(9000, IngressProtocol::Http, &[entry])]).is_err(),
                "{entry}"
            );
        }
        // Leading zeros inside an IPv6 hextet are ordinary notation.
        assert_eq!(
            normalize_allowlist(&["2001:0db8::/32".to_string()]).unwrap(),
            vec!["2001:db8::/32"]
        );
    }

    #[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!(info.deprecation.is_none());
        assert_eq!(info.error_message, None);
        assert_eq!(info.started_at, None);
    }

    #[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, ImageFilesystem, ImageSpec,
            PackageInstall, RunCommand,
        };
        let spec = ImageSpec {
            base: Some(BaseImage::Debian),
            architecture: ImageArchitecture::Arm64,
            python_version: "3.12".to_string(),
            filesystem: ImageFilesystem::Btrfs,
            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!(
            json.get("oci").is_none(),
            "unset oci arm must stay absent from the wire JSON"
        );
        assert_eq!(json["architecture"], json!("IMAGE_ARCHITECTURE_ARM64"));
        assert_eq!(json["pythonVersion"], json!("3.12"));
        assert_eq!(json["filesystem"], json!("IMAGE_FILESYSTEM_BTRFS"));
        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"));
    }

    // The oci arm of the source oneof renders as a nested message field,
    // exactly as protojson expects; the Go handler test decodes this shape.
    #[test]
    fn image_spec_serializes_oci_source_to_canonical_proto_json() {
        use crate::image::{ImageSpec, OciImage};
        let spec = ImageSpec {
            oci: Some(OciImage {
                reference: "docker.io/library/ubuntu@sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa".to_string(),
            }),
            ..Default::default()
        };
        let json = serde_json::to_value(&spec).unwrap();
        assert_eq!(
            json["oci"]["ref"],
            json!("docker.io/library/ubuntu@sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa")
        );
        assert!(json.get("base").is_none());
    }
}