sail/

imagebuild.rs

//! The custom-image build pipeline, shared by every SDK: resolve an
//! [`ImageDefinition`] (walk local directories, hash files, upload content)
//! into a content-addressed [`ImageSpec`], then build it to ready.
//!
//! The fluent builder DSL lives in each language wrapper; this module owns
//! everything below it (gitignore matching, bounds, hashing, presigned
//! uploads, the typed proto conversion, and the build poll loop) so the
//! wrappers stay thin and cannot drift.

use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::time::{Duration, Instant};

use futures::stream::{self, TryStreamExt};
use sha2::{Digest, Sha256};
use std::sync::Arc;

use crate::error::{SailError, TransportKind};
use crate::image::{
    AddLocalDirFile, BaseImage, ImageArchitecture, ImageBuildStep, ImageFilesystem, ImageSpec,
    OciImage, PackageInstall, RunCommand,
};
use crate::pb::image::v1 as pbimage;
use crate::pb::imagebuilder::v1 as pbimg;
use crate::Client;

/// S3's single-PUT ceiling; the backend enforces the same cap.
pub(crate) const MAX_LOCAL_FILE_BYTES: u64 = 5 * 1024 * 1024 * 1024;
/// Per-directory fail-fast bound, matching the backend.
pub(crate) const MAX_LOCAL_DIR_FILES: usize = 50_000;
/// Longest relative path allowed inside an uploaded directory, in bytes.
pub(crate) const MAX_LOCAL_DIR_RELATIVE_PATH_BYTES: usize = 1024;
/// Concurrent content uploads during a resolve.
const UPLOAD_CONCURRENCY: usize = 16;
/// Delay between build status polls.
const BUILD_POLL_INTERVAL: Duration = Duration::from_secs(1);
/// Floor for one presigned PUT, plus [`MIN_UPLOAD_BYTES_PER_SEC`] of body
/// budget: a stalled upload fails instead of hanging the resolve forever,
/// while a slow-but-progressing link keeps a generous allowance.
const UPLOAD_BASE_TIMEOUT: Duration = Duration::from_mins(5);
/// Throughput floor used to scale the upload budget with content size.
const MIN_UPLOAD_BYTES_PER_SEC: u64 = 1 << 20;
/// Transport-retry budget per imagebuilder RPC when no deadline bounds the
/// build (a deadline caps the budget at the time remaining instead).
const UNBOUNDED_BUILD_RPC_BUDGET: Duration = Duration::from_mins(1);

fn invalid(message: String) -> SailError {
    SailError::InvalidArgument { message }
}

/// Whether an image already built from the same definition satisfies a
/// build call, or the image is built again.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum BuildMode {
    /// Use the already-built image when one exists; build only when none
    /// does. Once an organization has a built image for the registry tag of
    /// an imported registry image ([`ImageDefinition::oci_ref`]), this keeps
    /// using the version the tag pointed to then, even if the tag has moved
    /// since.
    ReuseExisting,
    /// Build again even if a built image exists: a registry reference is
    /// looked up again (a moved tag is picked up) and the fresh build runs
    /// under a new image ID. Later builds and Sailbox creates of the same
    /// definition switch to the new image within a few minutes of it
    /// becoming ready, and keep using the previous image until then;
    /// existing Sailboxes are never changed. Forcing again while a forced
    /// rebuild is already running reuses that rebuild instead of starting
    /// another.
    ForceBuild,
}

/// The status of a custom image build.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ImageBuildStatus {
    /// The server reported a status this SDK version does not recognize.
    Unknown,
    /// Queued behind other builds.
    Queued,
    /// Building now.
    Building,
    /// Built and servable.
    Ready,
    /// The build failed; see the error message.
    Failed,
}

impl ImageBuildStatus {
    /// The wire string for this status.
    pub fn as_str(self) -> &'static str {
        match self {
            ImageBuildStatus::Unknown => "unknown",
            ImageBuildStatus::Queued => "queued",
            ImageBuildStatus::Building => "building",
            ImageBuildStatus::Ready => "ready",
            ImageBuildStatus::Failed => "failed",
        }
    }

    fn from_pb(status: i32) -> ImageBuildStatus {
        match pbimage::ImageBuildStatus::try_from(status) {
            Ok(pbimage::ImageBuildStatus::Queued) => ImageBuildStatus::Queued,
            Ok(pbimage::ImageBuildStatus::Building) => ImageBuildStatus::Building,
            Ok(pbimage::ImageBuildStatus::Ready) => ImageBuildStatus::Ready,
            Ok(pbimage::ImageBuildStatus::Failed) => ImageBuildStatus::Failed,
            _ => ImageBuildStatus::Unknown,
        }
    }
}

/// The state of a custom image build.
#[derive(Debug, Clone)]
#[non_exhaustive]
pub struct ImageBuild {
    /// The content-addressed image id.
    pub image_id: String,
    /// Build status.
    pub status: ImageBuildStatus,
    /// Human-readable failure detail when the status is failed, else empty.
    pub error_message: String,
    /// The digest-pinned form of the spec's registry reference when the
    /// spec's source is an OCI image, and empty otherwise. Creating from
    /// this reference instead of the submitted tag keeps naming the same
    /// registry content even if the tag has moved since.
    pub resolved_oci_ref: String,
}

/// The server's plan for uploading one content-addressed local file.
#[derive(Debug, Clone)]
pub(crate) enum LocalFileUploadPlan {
    /// The content is already stored; nothing to upload.
    AlreadyExists,
    /// Upload the bytes with one presigned PUT.
    SinglePart {
        /// The presigned URL to PUT to.
        upload_url: String,
        /// Headers the PUT must send.
        headers: HashMap<String, String>,
    },
}

/// One step of an [`ImageDefinition`]: a build operation, possibly referencing
/// local files that resolve uploads before the build.
#[derive(Debug, Clone)]
pub enum ImageDefinitionStep {
    /// Install system packages with apt.
    AptInstall(Vec<String>),
    /// Install Python packages with pip.
    PipInstall(Vec<String>),
    /// Run a shell command during the build.
    RunCommand(String),
    /// Bake one local file into the image.
    AddLocalFile {
        /// Path on this machine.
        local_path: PathBuf,
        /// Absolute POSIX path inside the image; a trailing `/` appends the
        /// source basename.
        remote_path: String,
        /// Permission bits (low 9); `None` uses the builder default (0644).
        mode: Option<u32>,
    },
    /// Bake a local directory tree into the image. Symlinks are skipped and
    /// file modes are preserved.
    AddLocalDir {
        /// Path on this machine.
        local_path: PathBuf,
        /// Absolute POSIX path of the directory root inside the image.
        remote_path: String,
        /// Gitignore-style patterns to skip.
        ignore: Vec<String>,
        /// A gitignore-style file whose patterns to skip (e.g. `.gitignore`).
        ignore_file: Option<PathBuf>,
    },
}

/// A custom image definition: a base image plus ordered build steps, where
/// local-file steps still reference paths on this machine. Resolve it with
/// [`Client::resolve_image`] (hash + upload) or hand it to
/// [`Client::build_image_definition`] to also build it to ready.
#[derive(Debug, Clone, Default)]
pub struct ImageDefinition {
    /// Base image to build on. Mutually exclusive with `oci_ref`.
    pub base: Option<BaseImage>,
    /// Your own image as the root filesystem: a reference to a Debian- or
    /// Ubuntu-based image whose first segment names a supported public
    /// registry (`docker.io`, `ghcr.io`, `public.ecr.aws`, or `quay.io`),
    /// with an optional `:tag` or `@sha256:<64 hex>` pin (no tag means the
    /// `latest` tag). Once Sail has a built image for a tag, later builds keep
    /// using it; if no image was ever built from a lookup, Sail may look the
    /// tag up again later. The image's `ENV` and `WORKDIR` become the Sailbox
    /// defaults for commands you run; its `ENTRYPOINT` and `CMD` are not run,
    /// because a Sailbox manages its own processes. Mutually exclusive with
    /// `base`.
    pub oci_ref: Option<String>,
    /// Target CPU architecture. Unspecified means amd64 with `base`, and with
    /// `oci_ref` means whichever architecture the registry image was built for
    /// (amd64 when it was built for both). Setting it with `oci_ref` requires
    /// the image to provide that architecture.
    pub architecture: ImageArchitecture,
    /// Environment variables baked into the image.
    pub env: HashMap<String, String>,
    /// Exact Python version to install as `python3`; empty uses the builder
    /// default. Not accepted with `oci_ref`: a registry image keeps its own
    /// `python3`.
    pub python_version: String,
    /// Writable root filesystem; unspecified preserves the ext4 default.
    pub filesystem: ImageFilesystem,
    /// Ordered build steps.
    pub steps: Vec<ImageDefinitionStep>,
}

/// Whether a spec is a bare builtin base the backend ships prebuilt (no build
/// needed): only build steps, env, or a pinned python version force a build.
/// A customer OCI source is never builtin — it always builds.
#[doc(hidden)]
pub fn is_builtin_base_spec(spec: &ImageSpec) -> bool {
    matches!(spec.base, Some(BaseImage::Debian | BaseImage::Devbox))
        && spec.oci.is_none()
        && spec.build_steps.is_empty()
        && spec.env.is_empty()
        && spec.python_version.is_empty()
        && matches!(
            spec.filesystem,
            ImageFilesystem::Unspecified | ImageFilesystem::Ext4
        )
}

/// Check that a customer OCI reference names a supported registry, so the
/// common mistakes fail before any hashing, uploading, or queueing. The
/// service parses the reference itself and stays authoritative on its shape.
/// Accepted forms are `name`, `name:tag`, and `name@sha256:<64 hex>`; a bare
/// name means the `latest` tag, and the backend resolves tags to digests at
/// submission.
pub(crate) fn validate_oci_ref(raw: &str) -> Result<(), SailError> {
    const MAX_OCI_REF_LENGTH: usize = 512;
    let reference = raw.trim();
    if reference.is_empty() {
        return Err(invalid("ociRef must be non-empty".to_string()));
    }
    if reference.len() > MAX_OCI_REF_LENGTH {
        return Err(invalid(format!(
            "ociRef exceeds {MAX_OCI_REF_LENGTH} characters"
        )));
    }
    // The registry is the reference's first path segment. A tag or digest can
    // only follow the first `/`, so splitting there isolates the registry
    // without parsing the rest. A reference with no `/` names no image;
    // Docker's normalizer would resolve it as a docker.io library repository.
    let Some((registry, repository)) = reference.split_once('/') else {
        return Err(invalid(format!(
            "ociRef {raw:?} must be fully qualified as registry/repository, e.g. docker.io/library/ubuntu:24.04"
        )));
    };
    if !ALLOWED_OCI_REGISTRIES.contains(&registry) {
        return Err(invalid(format!(
            "ociRef {raw:?} must name a supported public registry ({}) as its fully qualified first segment, e.g. docker.io/library/ubuntu:24.04",
            ALLOWED_OCI_REGISTRIES.join(", ")
        )));
    }
    // Docker Hub expands a single-segment repository into the implicit
    // `library` namespace (docker.io/ubuntu -> docker.io/library/ubuntu).
    // Accepting both spellings would map identical bytes to two image IDs, so
    // require the namespace to be written out.
    if registry == "docker.io" && !repository.contains('/') {
        return Err(invalid(format!(
            "ociRef {raw:?} must name the docker.io repository namespace, e.g. docker.io/library/ubuntu:24.04 for an official image"
        )));
    }
    Ok(())
}

/// Public registries an OCI base reference may name, matched against the
/// reference's first path segment. The service is authoritative; this
/// mirror only fails an unsupported registry fast, before any upload. Keep it
/// in step with the other SDKs and the service.
const ALLOWED_OCI_REGISTRIES: [&str; 4] = ["docker.io", "ghcr.io", "public.ecr.aws", "quay.io"];

/// Reject an image spec whose source is malformed before it reaches the wire.
/// The proto models the source as a oneof, so a spec that sets both a builtin
/// base and an OCI reference is ambiguous: the build path would send OCI while
/// `is_builtin_base_spec` classifies it from the base, and Sailbox creation
/// would serialize both and be rejected by the backend as a duplicate oneof
/// member. A directly constructed spec can also carry an unvalidated OCI
/// reference, so validate it here too. A spec with neither arm is the default
/// image and is valid.
pub(crate) fn validate_image_spec_source(spec: &ImageSpec) -> Result<(), SailError> {
    match (&spec.base, &spec.oci) {
        (Some(_), Some(_)) => Err(invalid(
            "an image takes either a builtin base or an OCI reference, not both".to_string(),
        )),
        (None, Some(oci)) => {
            if !spec.python_version.trim().is_empty() {
                return Err(invalid(
                    "a registry image keeps its own python3; pythonVersion is not supported with an OCI reference".to_string(),
                ));
            }
            validate_oci_ref(&oci.reference)
        }
        _ => Ok(()),
    }
}

/// Replace a spec's registry reference with the digest-pinned form a build
/// resolved, so whatever is created from the spec names exactly the built
/// bytes. A spec without an OCI source, or an empty resolution, is left
/// unchanged.
pub(crate) fn pin_resolved_oci_ref(spec: &mut ImageSpec, resolved_oci_ref: &str) {
    if resolved_oci_ref.is_empty() {
        return;
    }
    if let Some(oci) = spec.oci.as_mut() {
        oci.reference = resolved_oci_ref.to_string();
    }
}

/// Validate an in-image destination path: absolute POSIX, no `..`, no control
/// or shell-hostile characters, no trailing slash.
fn validate_remote_path(target: &str) -> Result<(), SailError> {
    if !target.starts_with('/') {
        return Err(invalid(format!("remotePath {target:?} must be absolute")));
    }
    if target.len() > 1 && target.ends_with('/') {
        return Err(invalid(format!(
            "remotePath {target:?} must not end with '/'"
        )));
    }
    for ch in target.chars() {
        let code = ch as u32;
        if code < 0x20 || code == 0x7f || matches!(ch, '"' | '\\' | '$' | ' ') {
            return Err(invalid(format!(
                "remotePath {target:?} contains an unsupported character"
            )));
        }
    }
    if target.split('/').any(|segment| segment == "..") {
        return Err(invalid(format!(
            "remotePath {target:?} must not contain '..'"
        )));
    }
    Ok(())
}

fn validate_mode(mode: Option<u32>) -> Result<u32, SailError> {
    match mode {
        None | Some(0) => Ok(0),
        Some(mode) if mode <= 0o777 => Ok(mode),
        Some(mode) => Err(invalid(format!(
            "mode 0o{mode:o} must fit in the low 9 bits"
        ))),
    }
}

/// Hash a local file with SHA-256, returning `(hex digest, size)`.
async fn hash_file(path: &Path) -> Result<(String, u64), SailError> {
    let path = path.to_path_buf();
    tokio::task::spawn_blocking(move || {
        use std::io::Read;
        let file = std::fs::File::open(&path)
            .map_err(|err| invalid(format!("cannot read {}: {err}", path.display())))?;
        let mut reader = std::io::BufReader::new(file);
        let mut hasher = Sha256::new();
        let mut buf = vec![0u8; 64 * 1024];
        let mut size: u64 = 0;
        loop {
            let n = reader
                .read(&mut buf)
                .map_err(|err| invalid(format!("cannot read {}: {err}", path.display())))?;
            if n == 0 {
                break;
            }
            hasher.update(&buf[..n]);
            size += n as u64;
        }
        Ok((format!("{:x}", hasher.finalize()), size))
    })
    .await
    .map_err(|err| SailError::Internal {
        message: format!("hashing task failed: {err}"),
    })?
}

struct WalkedFile {
    abs_path: PathBuf,
    relative_path: String,
    mode: u32,
}

/// Walk a local directory depth-first in sorted order, applying gitignore-style
/// matching, skipping symlinks, and enforcing the per-directory bounds.
fn walk_dir(
    root: &Path,
    matcher: &ignore::gitignore::Gitignore,
) -> Result<Vec<WalkedFile>, SailError> {
    fn recurse(
        root: &Path,
        dir: &Path,
        rel: &str,
        matcher: &ignore::gitignore::Gitignore,
        out: &mut Vec<WalkedFile>,
    ) -> Result<(), SailError> {
        let mut entries: Vec<_> = std::fs::read_dir(dir)
            .map_err(|err| invalid(format!("cannot read {}: {err}", dir.display())))?
            .collect::<Result<_, _>>()
            .map_err(|err| invalid(format!("cannot read {}: {err}", dir.display())))?;
        entries.sort_by_key(std::fs::DirEntry::file_name);
        for entry in entries {
            let name = entry
                .file_name()
                .to_str()
                .ok_or_else(|| {
                    invalid(format!(
                        "addLocalDir: {} has a non-UTF-8 file name",
                        entry.path().display()
                    ))
                })?
                .to_string();
            let rel_path = if rel.is_empty() {
                name.clone()
            } else {
                format!("{rel}/{name}")
            };
            let file_type = entry
                .file_type()
                .map_err(|err| invalid(format!("cannot stat {}: {err}", entry.path().display())))?;
            if file_type.is_symlink() {
                continue;
            }
            let is_dir = file_type.is_dir();
            if matcher
                .matched_path_or_any_parents(&rel_path, is_dir)
                .is_ignore()
            {
                continue;
            }
            if is_dir {
                recurse(root, &entry.path(), &rel_path, matcher, out)?;
                continue;
            }
            if !file_type.is_file() {
                continue;
            }
            if rel_path.len() > MAX_LOCAL_DIR_RELATIVE_PATH_BYTES {
                return Err(invalid(format!(
                    "relative path {rel_path} exceeds {MAX_LOCAL_DIR_RELATIVE_PATH_BYTES} bytes"
                )));
            }
            let metadata = entry
                .metadata()
                .map_err(|err| invalid(format!("cannot stat {}: {err}", entry.path().display())))?;
            if metadata.len() > MAX_LOCAL_FILE_BYTES {
                return Err(invalid(format!(
                    "{} ({} bytes) exceeds the {MAX_LOCAL_FILE_BYTES}-byte per-file limit",
                    entry.path().display(),
                    metadata.len()
                )));
            }
            out.push(WalkedFile {
                abs_path: entry.path(),
                relative_path: rel_path,
                mode: unix_mode(&metadata),
            });
            if out.len() > MAX_LOCAL_DIR_FILES {
                return Err(invalid(format!(
                    "{} has more than {MAX_LOCAL_DIR_FILES} files (max {MAX_LOCAL_DIR_FILES})",
                    root.display()
                )));
            }
        }
        Ok(())
    }

    let mut out = Vec::new();
    recurse(root, root, "", matcher, &mut out)?;
    Ok(out)
}

#[cfg(unix)]
fn unix_mode(metadata: &std::fs::Metadata) -> u32 {
    use std::os::unix::fs::PermissionsExt;
    metadata.permissions().mode() & 0o777
}

#[cfg(not(unix))]
fn unix_mode(_metadata: &std::fs::Metadata) -> u32 {
    0o644
}

fn ignore_matcher(
    root: &Path,
    patterns: &[String],
    ignore_file: Option<&Path>,
) -> Result<ignore::gitignore::Gitignore, SailError> {
    let mut builder = ignore::gitignore::GitignoreBuilder::new(root);
    if let Some(file) = ignore_file {
        if let Some(err) = builder.add(file) {
            return Err(invalid(format!(
                "cannot read ignore file {}: {err}",
                file.display()
            )));
        }
    }
    for pattern in patterns {
        builder
            .add_line(/* from */ None, pattern)
            .map_err(|err| invalid(format!("invalid ignore pattern {pattern:?}: {err}")))?;
    }
    builder
        .build()
        .map_err(|err| invalid(format!("invalid ignore patterns: {err}")))
}

// --- Typed proto conversion, shared by both bindings. ---

fn base_image_to_pb(base: BaseImage) -> pbimage::BaseImage {
    match base {
        BaseImage::Debian => pbimage::BaseImage::Debian,
        BaseImage::Devbox => pbimage::BaseImage::Devbox,
    }
}

fn architecture_to_pb(arch: ImageArchitecture) -> pbimage::ImageArchitecture {
    match arch {
        ImageArchitecture::Amd64 => pbimage::ImageArchitecture::Amd64,
        ImageArchitecture::Arm64 => pbimage::ImageArchitecture::Arm64,
        ImageArchitecture::Unspecified => pbimage::ImageArchitecture::Unspecified,
    }
}

fn filesystem_to_pb(filesystem: ImageFilesystem) -> pbimage::ImageFilesystem {
    match filesystem {
        ImageFilesystem::Unspecified => pbimage::ImageFilesystem::Unspecified,
        ImageFilesystem::Ext4 => pbimage::ImageFilesystem::Ext4,
        ImageFilesystem::Btrfs => pbimage::ImageFilesystem::Btrfs,
    }
}

fn build_step_to_pb(step: &ImageBuildStep) -> pbimage::ImageBuildStep {
    use pbimage::image_build_step::Step;
    let packages = |p: &PackageInstall| pbimage::PackageInstall {
        packages: p.packages.clone(),
    };
    let inner = match step {
        ImageBuildStep::AptInstall(p) => Step::AptInstall(packages(p)),
        ImageBuildStep::PipInstall(p) => Step::PipInstall(packages(p)),
        ImageBuildStep::RunCommand(c) => Step::RunCommand(pbimage::RunCommand {
            command: c.command.clone(),
        }),
        ImageBuildStep::AddLocalFile(f) => Step::AddLocalFile(pbimage::AddLocalFile {
            content_sha256: f.content_sha256.clone(),
            remote_path: f.remote_path.clone(),
            mode: f.mode,
        }),
        ImageBuildStep::AddLocalDir(d) => Step::AddLocalDir(pbimage::AddLocalDir {
            remote_path: d.remote_path.clone(),
            files: d
                .files
                .iter()
                .map(|file| pbimage::AddLocalDirFile {
                    relative_path: file.relative_path.clone(),
                    content_sha256: file.content_sha256.clone(),
                    mode: file.mode,
                })
                .collect(),
        }),
    };
    pbimage::ImageBuildStep { step: Some(inner) }
}

/// Convert a typed [`ImageSpec`] to its wire proto.
pub(crate) fn image_spec_to_pb(spec: &ImageSpec) -> pbimage::ImageSpec {
    let source = match (&spec.oci, spec.base) {
        (Some(oci), _) => Some(pbimage::image_spec::Source::Oci(pbimage::OciImage {
            r#ref: oci.reference.clone(),
        })),
        (None, Some(base)) => Some(pbimage::image_spec::Source::Base(
            base_image_to_pb(base) as i32
        )),
        (None, None) => None,
    };
    pbimage::ImageSpec {
        source,
        build_steps: spec.build_steps.iter().map(build_step_to_pb).collect(),
        env: spec.env.clone(),
        architecture: architecture_to_pb(spec.architecture) as i32,
        python_version: spec.python_version.clone(),
        filesystem: filesystem_to_pb(spec.filesystem) as i32,
    }
}

impl Client {
    /// The server's plan for uploading a content-addressed local file.
    pub(crate) async fn prepare_local_file_upload(
        &self,
        content_sha256: &str,
        content_length: u64,
    ) -> Result<LocalFileUploadPlan, SailError> {
        let request = pbimg::PrepareLocalFileUploadRequest {
            content_sha256: content_sha256.to_string(),
            content_length,
        };
        let response = self
            .imagebuilder()
            .prepare_local_file_upload(request)
            .await?;
        use pbimg::prepare_local_file_upload_response::Outcome;
        match response.outcome {
            Some(Outcome::AlreadyExists(_)) => Ok(LocalFileUploadPlan::AlreadyExists),
            Some(Outcome::SinglePart(plan)) => Ok(LocalFileUploadPlan::SinglePart {
                upload_url: plan.upload_url,
                headers: plan.required_headers,
            }),
            None => Err(SailError::Internal {
                message: "prepare_local_file_upload returned no outcome".to_string(),
            }),
        }
    }

    /// Submit or resume a custom image build. Poll
    /// [`Client::get_image_build_status`] until the status is ready or failed,
    /// or use [`Client::build_image_definition`] for the whole pipeline.
    ///
    /// `mode` selects whether an image already built for this spec satisfies
    /// the call or the image is built again; see [`BuildMode`].
    pub async fn build_image(
        &self,
        spec: &ImageSpec,
        retry_timeout_secs: f64,
        mode: BuildMode,
    ) -> Result<ImageBuild, SailError> {
        // Every build path funnels through here, so validating the source once
        // at this choke point rejects a both-arms or malformed-OCI spec before
        // the request crosses the wire, no matter which entry point (a direct
        // `build_image`, `build_spec_to_ready`, or `build_image_definition`
        // call) submitted it.
        validate_image_spec_source(spec)?;
        let request = pbimg::BuildImageRequest {
            image: Some(image_spec_to_pb(spec)),
            force_build: mode == BuildMode::ForceBuild,
        };
        let response = self
            .imagebuilder()
            .build_image(request, retry_timeout_secs)
            .await?;
        Ok(ImageBuild {
            image_id: response.image_id,
            status: ImageBuildStatus::from_pb(response.status),
            error_message: response.error_message,
            resolved_oci_ref: response.resolved_oci_ref,
        })
    }

    /// Poll one custom image build's status.
    pub async fn get_image_build_status(
        &self,
        image_id: &str,
        retry_timeout_secs: f64,
    ) -> Result<ImageBuild, SailError> {
        let request = pbimg::GetImageBuildStatusRequest {
            image_id: image_id.to_string(),
        };
        let response = self
            .imagebuilder()
            .get_image_build_status(request, retry_timeout_secs)
            .await?;
        Ok(ImageBuild {
            image_id: response.image_id,
            status: ImageBuildStatus::from_pb(response.status),
            error_message: response.error_message,
            resolved_oci_ref: response.resolved_oci_ref,
        })
    }

    /// Resolve one local file into a content-addressed `addLocalFile` step,
    /// uploading its bytes if the server does not already have them.
    #[doc(hidden)]
    pub async fn resolve_local_file_step(
        &self,
        local_path: &Path,
        remote_path: &str,
        mode: Option<u32>,
    ) -> Result<crate::image::AddLocalFile, SailError> {
        let metadata = std::fs::metadata(local_path).map_err(|_| {
            invalid(format!(
                "addLocalFile: {} does not exist or is not a file",
                local_path.display()
            ))
        })?;
        if !metadata.is_file() {
            return Err(invalid(format!(
                "addLocalFile: {} is not a file",
                local_path.display()
            )));
        }
        if metadata.len() > MAX_LOCAL_FILE_BYTES {
            return Err(invalid(format!(
                "addLocalFile: {} ({} bytes) exceeds the {MAX_LOCAL_FILE_BYTES}-byte limit",
                local_path.display(),
                metadata.len()
            )));
        }
        let mode = validate_mode(mode)?;
        let mut target = remote_path.to_string();
        if target.ends_with('/') {
            let basename = local_path
                .file_name()
                .map(|name| name.to_string_lossy().into_owned())
                .unwrap_or_default();
            target = format!("{target}{basename}");
        }
        validate_remote_path(&target)?;
        let (digest, size) = hash_file(local_path).await?;
        // A file still being written can grow past the stat-time check before
        // hashing finishes; the hash-time size is what actually uploads.
        if size > MAX_LOCAL_FILE_BYTES {
            return Err(invalid(format!(
                "addLocalFile: {} ({size} bytes) exceeds the {MAX_LOCAL_FILE_BYTES}-byte limit",
                local_path.display()
            )));
        }
        let http = reqwest::Client::new();
        self.upload_local_content(&http, &digest, local_path, size)
            .await?;
        Ok(crate::image::AddLocalFile {
            content_sha256: digest,
            remote_path: target,
            mode,
        })
    }

    /// Resolve one local directory into a content-addressed `addLocalDir`
    /// step: walk it with gitignore-style matching, hash every file, and
    /// upload content the server does not already have.
    #[doc(hidden)]
    pub async fn resolve_local_dir_step(
        &self,
        local_path: &Path,
        remote_path: &str,
        ignore: &[String],
        ignore_file: Option<&Path>,
    ) -> Result<crate::image::AddLocalDir, SailError> {
        let target = remote_path.trim_end_matches('/').to_string();
        if target.is_empty() {
            return Err(invalid(
                "addLocalDir: remotePath must not be '/'".to_string(),
            ));
        }
        validate_remote_path(&target)?;
        // The stat/walk phase is synchronous filesystem work that a large or
        // slow tree can stretch out; run it off the async runtime (like
        // hash_file) so the pipeline timeout can preempt it and other core
        // tasks keep running.
        let walk_root = local_path.to_path_buf();
        let ignore_owned = ignore.to_vec();
        let ignore_file_owned = ignore_file.map(Path::to_path_buf);
        let has_ignore = !ignore.is_empty() || ignore_file.is_some();
        let walked = tokio::task::spawn_blocking(move || {
            let metadata = std::fs::metadata(&walk_root).map_err(|_| {
                invalid(format!(
                    "addLocalDir: {} does not exist or is not a directory",
                    walk_root.display()
                ))
            })?;
            if !metadata.is_dir() {
                return Err(invalid(format!(
                    "addLocalDir: {} is not a directory",
                    walk_root.display()
                )));
            }
            let matcher = ignore_matcher(&walk_root, &ignore_owned, ignore_file_owned.as_deref())?;
            let walked = walk_dir(&walk_root, &matcher)?;
            if walked.is_empty() {
                let qualifier = if has_ignore {
                    " after applying ignore patterns"
                } else {
                    ""
                };
                return Err(invalid(format!(
                    "addLocalDir: {} contains no files{qualifier}",
                    walk_root.display()
                )));
            }
            Ok(walked)
        })
        .await
        .map_err(|err| SailError::Internal {
            message: format!("directory walk task failed: {err}"),
        })??;
        // digest -> (source path, size); deduped so shared content uploads once.
        let mut uploads: HashMap<String, (PathBuf, u64)> = HashMap::new();
        let mut files = Vec::with_capacity(walked.len());
        for file in walked {
            let (digest, size) = hash_file(&file.abs_path).await?;
            if size > MAX_LOCAL_FILE_BYTES {
                return Err(invalid(format!(
                    "addLocalDir: {} ({size} bytes) exceeds the {MAX_LOCAL_FILE_BYTES}-byte \
                     per-file limit",
                    file.abs_path.display()
                )));
            }
            uploads
                .entry(digest.clone())
                .or_insert_with(|| (file.abs_path.clone(), size));
            files.push(AddLocalDirFile {
                relative_path: file.relative_path,
                content_sha256: digest,
                mode: file.mode,
            });
        }
        files.sort_by(|a, b| a.relative_path.cmp(&b.relative_path));
        let http = reqwest::Client::new();
        stream::iter(uploads.into_iter().map(Ok::<_, SailError>))
            .try_for_each_concurrent(UPLOAD_CONCURRENCY, |(digest, (source, size))| {
                let http = http.clone();
                async move {
                    self.upload_local_content(&http, &digest, &source, size)
                        .await
                }
            })
            .await?;
        Ok(crate::image::AddLocalDir {
            remote_path: target,
            files,
        })
    }

    /// Resolve an [`ImageDefinition`] into a content-addressed [`ImageSpec`]:
    /// walk local directories, hash every file, and upload content the server
    /// does not already have.
    pub async fn resolve_image(&self, def: &ImageDefinition) -> Result<ImageSpec, SailError> {
        // Validate the image source before hashing or uploading any local
        // content, so an invalid OCI reference or a spec that sets both a
        // builtin base and an OCI source fails before those network side
        // effects. The probe carries the trimmed reference so the validator
        // sees the exact value the final spec will carry.
        let oci = def.oci_ref.as_deref().map(|raw| OciImage {
            reference: raw.trim().to_string(),
        });
        validate_image_spec_source(&ImageSpec {
            base: def.base,
            oci: oci.clone(),
            python_version: def.python_version.clone(),
            ..Default::default()
        })?;
        let mut steps = Vec::with_capacity(def.steps.len());
        for step in &def.steps {
            steps.push(match step {
                ImageDefinitionStep::AptInstall(packages) => {
                    ImageBuildStep::AptInstall(PackageInstall {
                        packages: packages.clone(),
                    })
                }
                ImageDefinitionStep::PipInstall(packages) => {
                    ImageBuildStep::PipInstall(PackageInstall {
                        packages: packages.clone(),
                    })
                }
                ImageDefinitionStep::RunCommand(command) => {
                    ImageBuildStep::RunCommand(RunCommand {
                        command: command.clone(),
                    })
                }
                ImageDefinitionStep::AddLocalFile {
                    local_path,
                    remote_path,
                    mode,
                } => ImageBuildStep::AddLocalFile(
                    self.resolve_local_file_step(local_path, remote_path, *mode)
                        .await?,
                ),
                ImageDefinitionStep::AddLocalDir {
                    local_path,
                    remote_path,
                    ignore,
                    ignore_file,
                } => ImageBuildStep::AddLocalDir(
                    self.resolve_local_dir_step(
                        local_path,
                        remote_path,
                        ignore,
                        ignore_file.as_deref(),
                    )
                    .await?,
                ),
            });
        }
        Ok(ImageSpec {
            base: def.base,
            oci,
            build_steps: steps,
            env: def.env.clone(),
            architecture: def.architecture,
            python_version: def.python_version.clone(),
            filesystem: def.filesystem,
        })
    }

    /// Upload one content-addressed local file if the server does not already
    /// have it.
    async fn upload_local_content(
        &self,
        http: &reqwest::Client,
        digest: &str,
        source: &Path,
        size: u64,
    ) -> Result<(), SailError> {
        let plan = self.prepare_local_file_upload(digest, size).await?;
        let LocalFileUploadPlan::SinglePart {
            upload_url,
            headers,
        } = plan
        else {
            return Ok(());
        };
        let file = tokio::fs::File::open(source)
            .await
            .map_err(|err| invalid(format!("cannot read {}: {err}", source.display())))?;
        let (request, streamed_digest) = sized_put_request(http, &upload_url, file, size, &headers);
        let response = tokio::time::timeout(upload_timeout(size), request.send())
            .await
            .map_err(|_| SailError::Transport {
                kind: TransportKind::Timeout,
                message: format!("local file upload stalled ({size} bytes not delivered in time)"),
                source: None,
            })?
            .map_err(|err| SailError::Transport {
                kind: TransportKind::Connection,
                message: format!("local file upload failed: {err}"),
                source: None,
            })?;
        if !response.status().is_success() {
            return Err(SailError::Api {
                message: format!(
                    "local file upload failed: HTTP {} {}",
                    response.status().as_u16(),
                    response.status().canonical_reason().unwrap_or("")
                ),
                status: response.status().as_u16(),
                body: serde_json::Value::Null,
            });
        }
        // The file was hashed before this second open; a rewrite in between
        // (same size, different bytes) would poison the content-addressed
        // store under the old digest. The body hashed what it actually
        // streamed, so fail the build instead of using a mismatched object.
        let streamed = streamed_digest.lock().unwrap().take();
        if streamed.as_deref() != Some(digest) {
            return Err(invalid(format!(
                "{} changed while it was being uploaded; retry the build",
                source.display()
            )));
        }
        Ok(())
    }

    /// Build an already-resolved spec to ready, bounded by `timeout` (an
    /// unrepresentably large value waits indefinitely). The envelope both
    /// bridges and [`Client::build_image_definition`] share. Readiness is
    /// memoized per client (see [`crate::imagecache`]): concurrent callers
    /// share one build, a completed build serves later callers until the
    /// refresh window lapses, and failures always retry.
    /// [`BuildMode::ForceBuild`] skips that memoization and starts a fresh
    /// build on the server.
    #[doc(hidden)]
    pub async fn build_spec_with_timeout(
        &self,
        spec: &ImageSpec,
        timeout: Duration,
        mode: BuildMode,
    ) -> Result<ImageBuild, SailError> {
        match Instant::now().checked_add(timeout) {
            None => {
                self.build_spec_ready_cached(spec, timeout, /* recovery */ false, mode)
                    .await
            }
            Some(_) => tokio::time::timeout(
                timeout,
                self.build_spec_ready_cached(spec, timeout, /* recovery */ false, mode),
            )
            .await
            .unwrap_or_else(|_| {
                Err(SailError::Transport {
                    kind: TransportKind::Timeout,
                    message: "timed out building the image".to_string(),
                    source: None,
                })
            }),
        }
    }

    /// Build a spec to ready through the client's readiness cache. Callers
    /// share one build per spec whatever `timeout` each passes: a build still
    /// running is joined, and a joiner inherits that build's deadline. A
    /// joiner that saw the joined build hit its deadline retries with a fresh
    /// entry, so joining never shortens the caller's own budget (the caller's
    /// outer envelope still bounds the total wait).
    pub(crate) async fn build_spec_ready_cached(
        &self,
        spec: &ImageSpec,
        timeout: Duration,
        recovery: bool,
        mode: BuildMode,
    ) -> Result<ImageBuild, SailError> {
        let key = canonical_spec_key(spec)?;
        // Other clients can force the organization's resolution of a tag to
        // move. Share a tag build while it is running, but do not let this
        // client's settled result hide that external change. Immutable digest
        // references and non-OCI specs keep the readiness optimization.
        let retain_ready = match &spec.oci {
            Some(oci) => oci.reference.contains("@sha256:"),
            None => true,
        };
        loop {
            let joined = self.image_ready_cache().join_or_lead(
                &key,
                recovery,
                mode == BuildMode::ForceBuild,
                |id| {
                    let client = self.clone();
                    let spec = spec.clone();
                    let key = key.clone();
                    let deadline = Instant::now().checked_add(timeout);
                    futures::FutureExt::shared(futures::FutureExt::boxed(async move {
                        let result = client
                            .build_spec_to_ready_inner(&spec, deadline, mode)
                            .await;
                        match &result {
                            Ok(build) => {
                                client.image_ready_cache().settle_success(
                                    &key,
                                    id,
                                    build.clone(),
                                    retain_ready,
                                );
                            }
                            Err(_) => client.image_ready_cache().settle_failure(&key, id),
                        }
                        result.map_err(Arc::new)
                    }))
                },
            );
            let (shared, led) = match joined {
                crate::imagecache::Joined::Ready(build) => return Ok(build),
                crate::imagecache::Joined::Pending { build, led } => (build, led),
            };
            match shared.await {
                Ok(build) => return Ok(build),
                Err(err) => {
                    let timed_out = matches!(
                        err.as_ref(),
                        SailError::Transport {
                            kind: TransportKind::Timeout,
                            ..
                        }
                    );
                    if led || !timed_out {
                        // A sole caller (the common case) unwraps the original
                        // error; concurrent failure waiters each get a copy
                        // whose source chains to the shared original.
                        return Err(
                            Arc::try_unwrap(err).unwrap_or_else(|arc| SailError::fan_out(&arc))
                        );
                    }
                }
            }
        }
    }

    /// Resolve an [`ImageDefinition`] and build it to ready, returning the
    /// content-addressed [`ImageSpec`] to create Sailboxes from. A bare
    /// builtin base skips the build. `timeout` bounds the whole pipeline
    /// (hashing, uploads, and the build); 30 minutes is a good default, and
    /// [`Duration::MAX`] waits indefinitely. Local files are re-hashed on
    /// every call, so edits always reach the build, and rebuilding an
    /// unchanged, already-built image returns quickly. For an image imported
    /// from a registry ([`ImageDefinition::oci_ref`]), the returned spec is
    /// pinned to the exact registry version the build used, so Sailboxes
    /// created from it get those bytes even if the tag moves later.
    ///
    /// `mode` selects whether the image already built for this definition
    /// satisfies the call or the image is built again; see [`BuildMode`].
    pub async fn build_image_definition(
        &self,
        def: &ImageDefinition,
        timeout: Duration,
        mode: BuildMode,
    ) -> Result<ImageSpec, SailError> {
        let work = async {
            let mut spec = self.resolve_image(def).await?;
            if is_builtin_base_spec(&spec) {
                return Ok(spec);
            }
            let build = self
                .build_spec_ready_cached(&spec, timeout, /* recovery */ false, mode)
                .await?;
            // The returned spec is what callers create Sailboxes from; pin it
            // to the reference the build resolved so those creates name the
            // built bytes even if a tag has moved since.
            pin_resolved_oci_ref(&mut spec, &build.resolved_oci_ref);
            Ok(spec)
        };
        match Instant::now().checked_add(timeout) {
            None => work.await,
            Some(_) => tokio::time::timeout(timeout, work)
                .await
                .unwrap_or_else(|_| {
                    Err(SailError::Transport {
                        kind: TransportKind::Timeout,
                        message: "timed out building the image".to_string(),
                        source: None,
                    })
                }),
        }
    }

    /// Build an already-resolved spec to ready (submit + poll).
    #[doc(hidden)]
    pub async fn build_spec_to_ready(
        &self,
        spec: &ImageSpec,
        deadline: Option<Instant>,
    ) -> Result<ImageBuild, SailError> {
        self.build_spec_to_ready_inner(spec, deadline, BuildMode::ReuseExisting)
            .await
    }

    async fn build_spec_to_ready_inner(
        &self,
        spec: &ImageSpec,
        deadline: Option<Instant>,
        mode: BuildMode,
    ) -> Result<ImageBuild, SailError> {
        // Per-RPC transport-retry budget: the time left until the deadline,
        // or a fixed bound when the caller waits indefinitely.
        let rpc_budget = || {
            deadline.map_or(UNBOUNDED_BUILD_RPC_BUDGET.as_secs_f64(), |deadline| {
                deadline
                    .saturating_duration_since(Instant::now())
                    .as_secs_f64()
            })
        };
        let mut build = self.build_image(spec, rpc_budget(), mode).await?;
        loop {
            match build.status {
                ImageBuildStatus::Ready => return Ok(build),
                ImageBuildStatus::Failed => {
                    let message = if build.error_message.is_empty() {
                        "image build failed".to_string()
                    } else {
                        build.error_message.clone()
                    };
                    return Err(SailError::ImageBuild { message });
                }
                _ => {}
            }
            let nap = match deadline {
                None => BUILD_POLL_INTERVAL,
                Some(deadline) => {
                    let left = deadline.saturating_duration_since(Instant::now());
                    if left.is_zero() {
                        return Err(SailError::Transport {
                            kind: TransportKind::Timeout,
                            message: format!(
                                "timed out waiting for image build {}",
                                build.image_id
                            ),
                            source: None,
                        });
                    }
                    left.min(BUILD_POLL_INTERVAL)
                }
            };
            tokio::time::sleep(nap).await;
            build = self
                .get_image_build_status(&build.image_id, rpc_budget())
                .await?;
        }
    }
}

/// The readiness-cache identity of a spec: the sha256 of its canonical
/// (key-sorted) JSON, the same serialization the create request sends.
/// Hashing bounds key memory for specs carrying many content digests.
///
/// The sort is applied here rather than inherited from serde_json's default
/// `Map` being a `BTreeMap`. `ImageSpec::env` is a `HashMap`, whose iteration
/// order is seeded per process, so under a build where object order is
/// insertion order (serde_json's `preserve_order`, which feature unification can
/// switch on from anywhere in the workspace) the same spec would hash
/// differently in two CLI invocations, splitting the cache and rebuilding images
/// that were already ready. Sorting explicitly reproduces the historical keys
/// exactly, so no cache is invalidated by making this guarantee our own.
pub(crate) fn canonical_spec_key(spec: &ImageSpec) -> Result<String, SailError> {
    let value = serde_json::to_value(spec).map_err(|err| SailError::Internal {
        message: format!("serialize image spec: {err}"),
    })?;
    let mut hasher = Sha256::new();
    hasher.update(sorted_json(&value).to_string().as_bytes());
    Ok(format!("{:x}", hasher.finalize()))
}

/// Rebuild `value` with every object's keys in sorted order.
fn sorted_json(value: &serde_json::Value) -> serde_json::Value {
    match value {
        serde_json::Value::Object(map) => {
            let mut keys: Vec<&String> = map.keys().collect();
            keys.sort();
            let mut sorted = serde_json::Map::with_capacity(map.len());
            for key in keys {
                sorted.insert(key.clone(), sorted_json(&map[key]));
            }
            serde_json::Value::Object(sorted)
        }
        serde_json::Value::Array(items) => {
            serde_json::Value::Array(items.iter().map(sorted_json).collect())
        }
        other => other.clone(),
    }
}

/// Build the presigned PUT for one content-addressed upload. Presigned PUT
/// endpoints reject chunked transfer encoding, so the body must advertise its
/// exact size; hyper then frames the request with Content-Length while the
/// file still streams from disk.
fn sized_put_request(
    http: &reqwest::Client,
    upload_url: &str,
    file: tokio::fs::File,
    size: u64,
    headers: &HashMap<String, String>,
) -> (
    reqwest::RequestBuilder,
    Arc<std::sync::Mutex<Option<String>>>,
) {
    let (body, streamed_digest) = SizedFileBody::new(file, size);
    let mut request = http.put(upload_url).body(reqwest::Body::wrap(body));
    for (name, value) in headers {
        request = request.header(name, value);
    }
    (request, streamed_digest)
}

/// The whole-request budget for one presigned PUT: a base allowance plus the
/// body at a conservative throughput floor.
fn upload_timeout(size: u64) -> Duration {
    UPLOAD_BASE_TIMEOUT + Duration::from_secs(size / MIN_UPLOAD_BYTES_PER_SEC)
}

/// A streaming request body over a file with an exact size hint. Presigned
/// PUT endpoints reject chunked transfer encoding, so the body must report
/// its length up front; the file itself still streams from disk in 64 KiB
/// frames rather than being buffered whole.
struct SizedFileBody {
    reader: tokio_util::io::ReaderStream<tokio::fs::File>,
    remaining: u64,
    hasher: Option<sha2::Sha256>,
    streamed_digest: Arc<std::sync::Mutex<Option<String>>>,
}

impl SizedFileBody {
    fn new(file: tokio::fs::File, size: u64) -> (Self, Arc<std::sync::Mutex<Option<String>>>) {
        let streamed_digest = Arc::new(std::sync::Mutex::new(None));
        let mut hasher = Some(sha2::Sha256::new());
        if size == 0 {
            // An empty body may never be polled; its digest is already known.
            *streamed_digest.lock().unwrap() =
                Some(format!("{:x}", hasher.take().unwrap().finalize()));
        }
        (
            SizedFileBody {
                reader: tokio_util::io::ReaderStream::new(file),
                remaining: size,
                hasher,
                streamed_digest: Arc::clone(&streamed_digest),
            },
            streamed_digest,
        )
    }
}

impl http_body::Body for SizedFileBody {
    type Data = bytes::Bytes;
    type Error = std::io::Error;

    fn poll_frame(
        mut self: std::pin::Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
    ) -> std::task::Poll<Option<Result<http_body::Frame<Self::Data>, Self::Error>>> {
        use futures::Stream;
        match std::pin::Pin::new(&mut self.reader).poll_next(cx) {
            std::task::Poll::Ready(Some(Ok(chunk))) => {
                self.remaining = self.remaining.saturating_sub(chunk.len() as u64);
                if let Some(hasher) = self.hasher.as_mut() {
                    hasher.update(&chunk);
                }
                // Exact Content-Length framing means the final end-of-stream
                // poll may never come; finalize as soon as the advertised
                // bytes have been streamed.
                if self.remaining == 0 {
                    if let Some(hasher) = self.hasher.take() {
                        *self.streamed_digest.lock().unwrap() =
                            Some(format!("{:x}", hasher.finalize()));
                    }
                }
                std::task::Poll::Ready(Some(Ok(http_body::Frame::data(chunk))))
            }
            std::task::Poll::Ready(Some(Err(err))) => std::task::Poll::Ready(Some(Err(err))),
            std::task::Poll::Ready(None) => {
                if let Some(hasher) = self.hasher.take() {
                    *self.streamed_digest.lock().unwrap() =
                        Some(format!("{:x}", hasher.finalize()));
                }
                std::task::Poll::Ready(None)
            }
            std::task::Poll::Pending => std::task::Poll::Pending,
        }
    }

    fn is_end_stream(&self) -> bool {
        self.remaining == 0
    }

    fn size_hint(&self) -> http_body::SizeHint {
        http_body::SizeHint::with_exact(self.remaining)
    }
}

#[cfg(test)]
mod tests {
    #[test]
    fn pin_resolved_oci_ref_replaces_only_an_oci_source() {
        use crate::image::{BaseImage, ImageSpec, OciImage};
        let digest = format!("docker.io/library/python@sha256:{}", "a".repeat(64));
        let mut oci = ImageSpec {
            oci: Some(OciImage {
                reference: "docker.io/library/python:3.13".to_string(),
            }),
            ..Default::default()
        };
        super::pin_resolved_oci_ref(&mut oci, &digest);
        assert_eq!(oci.oci.unwrap().reference, digest);
        // An empty resolution (a builtin-base build) changes nothing.
        let mut unresolved = ImageSpec {
            oci: Some(OciImage {
                reference: "docker.io/library/python:3.13".to_string(),
            }),
            ..Default::default()
        };
        super::pin_resolved_oci_ref(&mut unresolved, "");
        assert_eq!(
            unresolved.oci.unwrap().reference,
            "docker.io/library/python:3.13"
        );
        // A base spec has no reference to pin.
        let mut base = ImageSpec {
            base: Some(BaseImage::Debian),
            ..Default::default()
        };
        super::pin_resolved_oci_ref(&mut base, &digest);
        assert!(base.oci.is_none());
    }

    #[test]
    fn oci_ref_validation() {
        const DIGEST: &str =
            "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa";
        for good in [
            format!("docker.io/library/ubuntu@{DIGEST}"),
            format!("ghcr.io/acme/my-tool@{DIGEST}"),
            format!("public.ecr.aws/lts/ubuntu@{DIGEST}"),
            format!("quay.io/org/base@{DIGEST}"),
            format!("  docker.io/library/ubuntu@{DIGEST}  "),
            // Tags are accepted, and so is a bare name, which means the
            // `latest` tag; the backend pins them to a digest at submission.
            "docker.io/library/ubuntu:24.04".to_string(),
            "docker.io/library/ubuntu".to_string(),
            "ghcr.io/acme/my-tool:v1.2.3-RC1".to_string(),
            format!("ghcr.io/acme/build--tools@{DIGEST}"),
        ] {
            super::validate_oci_ref(&good).unwrap_or_else(|err| panic!("{good:?} rejected: {err}"));
        }
        // The client checks the registry, not the shape of the reference: the
        // backend parses references and rejects malformed ones at submission.
        for bad in [
            String::new(),
            // Unqualified Docker Hub shorthand, with and without a tag: it
            // carries no registry or repository path, so Docker would
            // normalize it to a docker.io library repository.
            "ubuntu:24.04".to_string(),
            "ubuntu".to_string(),
            format!("ubuntu@{DIGEST}"),
            // A bare registry names no image either.
            format!("ghcr.io@{DIGEST}"),
            "docker.io".to_string(),
            // Well-formed but disallowed registries: a private address, an
            // internal host, a host with a port, a public-but-unlisted
            // registry, and a lookalike that merely starts with an allowed
            // name. Each must fail closed.
            format!("10.0.0.1/repo@{DIGEST}"),
            format!("registry.internal/repo@{DIGEST}"),
            format!("localhost:5000/repo@{DIGEST}"),
            format!("gcr.io/library/ubuntu@{DIGEST}"),
            format!("docker.io.evil.example/repo@{DIGEST}"),
            // docker.io/ubuntu is a single-segment repository Docker Hub
            // expands to docker.io/library/ubuntu; both spellings would map
            // identical bytes to two image IDs, so require the namespace.
            format!("docker.io/ubuntu@{DIGEST}"),
        ] {
            assert!(
                super::validate_oci_ref(&bad).is_err(),
                "{bad:?} unexpectedly accepted"
            );
        }
    }

    #[test]
    fn oci_spec_is_never_builtin_and_maps_to_the_oci_oneof_arm() {
        use crate::image::{ImageSpec, OciImage};
        let spec = ImageSpec {
            oci: Some(OciImage {
                reference:
                    "ubuntu@sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
                        .to_string(),
            }),
            ..Default::default()
        };
        assert!(!super::is_builtin_base_spec(&spec));
        let pb = super::image_spec_to_pb(&spec);
        match pb.source {
            Some(crate::pb::image::v1::image_spec::Source::Oci(oci)) => {
                assert_eq!(oci.r#ref, spec.oci.as_ref().unwrap().reference);
            }
            other => panic!("pb source = {other:?}, want the oci arm"),
        }
    }

    #[test]
    fn image_spec_source_rejects_both_arms_and_bad_oci() {
        use crate::image::{BaseImage, ImageSpec, OciImage};
        const DIGEST: &str =
            "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa";
        // Both a builtin base and an OCI source is ambiguous; the backend
        // rejects it, so the client must too.
        let both = ImageSpec {
            base: Some(BaseImage::Debian),
            oci: Some(OciImage {
                reference: format!("docker.io/library/ubuntu@{DIGEST}"),
            }),
            ..Default::default()
        };
        assert!(!super::is_builtin_base_spec(&both));
        assert!(super::validate_image_spec_source(&both).is_err());
        // An OCI-only spec still has its reference validated here.
        let bad_oci = ImageSpec {
            oci: Some(OciImage {
                reference: "ubuntu:24.04".to_string(),
            }),
            ..Default::default()
        };
        assert!(super::validate_image_spec_source(&bad_oci).is_err());
        // The server rejects python_version with an OCI source (the image
        // keeps its own python3); mirroring it here fails the spec before
        // local files are hashed and uploaded. No language wrapper can build
        // this pairing, so this chokepoint is its only client-side check.
        let pinned_python = ImageSpec {
            oci: Some(OciImage {
                reference: format!("docker.io/library/ubuntu@{DIGEST}"),
            }),
            python_version: "3.12.13".to_string(),
            ..Default::default()
        };
        assert!(super::validate_image_spec_source(&pinned_python).is_err());
        // A base-only spec, the default (no source), and a well-formed OCI-only
        // spec all pass.
        let base_only = ImageSpec {
            base: Some(BaseImage::Debian),
            ..Default::default()
        };
        assert!(super::validate_image_spec_source(&base_only).is_ok());
        assert!(super::validate_image_spec_source(&ImageSpec::default()).is_ok());
        let good_oci = ImageSpec {
            oci: Some(OciImage {
                reference: format!("docker.io/library/ubuntu@{DIGEST}"),
            }),
            ..Default::default()
        };
        assert!(super::validate_image_spec_source(&good_oci).is_ok());
    }

    #[test]
    fn upload_budget_scales_with_content_size() {
        assert_eq!(upload_timeout(0), Duration::from_mins(5));
        // 1 GiB at the 1 MiB/s floor adds 1024s to the base allowance.
        assert_eq!(
            upload_timeout(1 << 30),
            Duration::from_mins(5) + Duration::from_secs(1024)
        );
    }

    #[tokio::test]
    async fn upload_body_advertises_its_exact_size() {
        // The presigned plan's endpoint rejects chunked transfer encoding.
        // Framing is decided from the body's own size hint (a manual
        // Content-Length header is not sufficient on every protocol), so the
        // body must report the exact size before any bytes are read.
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("payload.bin");
        std::fs::write(&path, b"0123456789").expect("write");
        let file = tokio::fs::File::open(&path).await.expect("open");
        let (body, _digest) = SizedFileBody::new(file, 10);
        assert_eq!(http_body::Body::size_hint(&body).exact(), Some(10));
        assert!(!http_body::Body::is_end_stream(&body));
    }

    #[tokio::test]
    async fn presigned_put_uses_content_length_framing() {
        use tokio::io::{AsyncReadExt, AsyncWriteExt};

        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("payload.bin");
        std::fs::write(&path, b"0123456789").expect("write");

        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
            .await
            .expect("bind");
        let addr = listener.local_addr().expect("addr");
        let server = tokio::spawn(async move {
            let (mut sock, _) = listener.accept().await.expect("accept");
            let mut raw = Vec::new();
            let mut buf = [0u8; 4096];
            loop {
                let n = sock.read(&mut buf).await.expect("read");
                raw.extend_from_slice(&buf[..n]);
                if let Some(head_end) = raw.windows(4).position(|w| w == b"\r\n\r\n") {
                    let head = String::from_utf8_lossy(&raw[..head_end]).to_lowercase();
                    let body_len = raw.len() - (head_end + 4);
                    if body_len >= 10 {
                        sock.write_all(b"HTTP/1.1 200 OK\r\ncontent-length: 0\r\n\r\n")
                            .await
                            .expect("respond");
                        return head;
                    }
                }
            }
        });

        let file = tokio::fs::File::open(&path).await.expect("open");
        let headers = HashMap::from([(
            "Content-Type".to_string(),
            "application/octet-stream".to_string(),
        )]);
        let (request, streamed_digest) = sized_put_request(
            &reqwest::Client::new(),
            &format!("http://{addr}/upload"),
            file,
            10,
            &headers,
        );
        let response = request.send().await.expect("send");
        assert!(response.status().is_success());
        // The body hashed exactly what it streamed.
        assert_eq!(
            streamed_digest.lock().unwrap().as_deref(),
            Some("84d89877f0d4041efb6bf91a16f0248f2fd573e6af05c19f96bedb9f882f7882")
        );

        let head = server.await.expect("server");
        // Presigned endpoints reject chunked transfer encoding; the request
        // must carry the exact Content-Length instead.
        assert!(
            head.contains("content-length: 10"),
            "missing sized framing in request head: {head}"
        );
        assert!(
            !head.contains("transfer-encoding"),
            "request must not be chunked: {head}"
        );
    }

    use super::*;

    #[test]
    fn btrfs_base_requires_a_build_while_ext4_keeps_the_builtin_fast_path() {
        let base = ImageSpec {
            base: Some(BaseImage::Debian),
            ..Default::default()
        };
        assert!(is_builtin_base_spec(&base));

        let explicit_ext4 = ImageSpec {
            filesystem: ImageFilesystem::Ext4,
            ..base.clone()
        };
        assert!(is_builtin_base_spec(&explicit_ext4));

        let btrfs = ImageSpec {
            filesystem: ImageFilesystem::Btrfs,
            ..base
        };
        assert!(!is_builtin_base_spec(&btrfs));
        assert_eq!(
            image_spec_to_pb(&btrfs).filesystem,
            pbimage::ImageFilesystem::Btrfs as i32
        );
    }

    #[test]
    fn remote_path_rules_match_the_wrappers() {
        assert!(validate_remote_path("/app/config.json").is_ok());
        assert!(validate_remote_path("relative").is_err());
        assert!(validate_remote_path("/app/").is_err());
        assert!(validate_remote_path("/app/../etc").is_err());
        assert!(validate_remote_path("/app/with space").is_err());
        assert!(validate_remote_path("/app/$HOME").is_err());
        assert!(validate_mode(Some(0o600)).is_ok());
        assert!(validate_mode(Some(0o1777)).is_err());
    }

    #[tokio::test]
    async fn resolve_walks_hashes_and_respects_gitignore() {
        let dir = tempfile::tempdir().expect("tempdir");
        std::fs::create_dir_all(dir.path().join("src/generated")).unwrap();
        std::fs::write(dir.path().join("src/keep.py"), b"keep").unwrap();
        std::fs::write(dir.path().join("src/skip.pyc"), b"skip").unwrap();
        std::fs::write(dir.path().join("src/generated/gen.py"), b"gen").unwrap();
        std::fs::write(dir.path().join("top.txt"), b"top").unwrap();

        let matcher = ignore_matcher(
            dir.path(),
            &["*.pyc".to_string(), "src/generated/".to_string()],
            /* ignore_file */ None,
        )
        .expect("matcher");
        let walked = walk_dir(dir.path(), &matcher).expect("walk");
        let mut paths: Vec<_> = walked.iter().map(|f| f.relative_path.clone()).collect();
        paths.sort();
        assert_eq!(paths, ["src/keep.py", "top.txt"]);

        let (digest, size) = hash_file(&dir.path().join("top.txt")).await.expect("hash");
        assert_eq!(size, 3);
        assert_eq!(
            digest,
            "28720365c5e7476a011e4f43ac003ee5f16247a263b9d623aa85ed311d73bf39"
        );
    }
}