sail-rs 0.7.1

//! 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, DockerfileFromResolution, ImageArchitecture, ImageBuildStep,
    ImageFilesystem, ImageSpec, OciImage, PackageInstall, RunCommand,
};
use crate::imagecache::BuildOrigin;
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;
/// Largest Dockerfile text accepted; the backend enforces the same cap.
pub(crate) const MAX_DOCKERFILE_BYTES: usize = 512 * 1024;
/// Most files a Dockerfile build context may name, matching the backend
/// (deliberately below [`MAX_LOCAL_DIR_FILES`]: the manifest travels inline
/// in the build request).
pub(crate) const MAX_DOCKERFILE_CONTEXT_FILES: usize = 10_000;
/// Most build args a Dockerfile build accepts, matching the backend.
pub(crate) const MAX_DOCKERFILE_BUILD_ARGS: usize = 64;
/// Longest build-arg key in bytes, matching the backend.
pub(crate) const MAX_DOCKERFILE_BUILD_ARG_KEY_BYTES: usize = 128;
/// Longest build-arg value in bytes, matching the backend.
pub(crate) const MAX_DOCKERFILE_BUILD_ARG_VALUE_BYTES: usize = 4096;
/// 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);
// Keep in sync with imagebuilder.GuestSchemaSupersededMessage. Older service
// pods cannot populate the typed retryable field during a rolling deployment,
// so the shared SDK core recognizes this one stable message as a compatibility
// bridge. New service pods set the typed field.
const GUEST_SCHEMA_SUPERSEDED_MESSAGE: &str =
    "image build did not complete; submit the build again";

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. The tags a Dockerfile's `FROM` and `COPY --from` instructions
    /// name ([`ImageDefinition::dockerfile`]) hold to their first-use
    /// versions the same way.
    ReuseExisting,
    /// Build again even if a built image exists: the fresh build runs under
    /// a new image ID, and this call waits for it to become ready. New
    /// Sailboxes use the fresh image once it is ready, Sailboxes that
    /// already exist keep the filesystem they were created with, and a
    /// forced build that fails changes nothing. For an image imported
    /// through a registry tag ([`ImageDefinition::oci_ref`]), a forced
    /// build also asks the registry what the tag points at now and builds
    /// that version. The tag then means that version for your whole
    /// organization, while specs built earlier keep their pinned version.
    /// For an image built from a Dockerfile
    /// ([`ImageDefinition::dockerfile`]), a forced build looks up the tags
    /// its `FROM` and `COPY --from` instructions name and moves those pins
    /// for your whole organization, while specs built earlier keep the
    /// versions their build used. If
    /// forced builds overlap, the last-requested one that succeeds decides
    /// which image new Sailboxes use and, for a tag, what the tag means.
    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,
    // Whether the high-level build loop can safely submit the original spec
    // again. This is not a status callers should need to handle.
    pub(crate) retryable: bool,
    /// 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,
    /// What each external image reference resolved to when the spec's
    /// source is a Dockerfile, and `None` otherwise. Carrying these in the
    /// spec's `pinned_from` keeps creating from the image this build
    /// produced, even after a forced build moves what the references mean
    /// for your organization.
    pub dockerfile_pins: Option<Vec<DockerfileFromResolution>>,
}

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

/// The Dockerfile of a [`DockerfileSource`]: a path to read or literal text.
#[derive(Debug, Clone)]
pub enum DockerfileInput {
    /// Path to a Dockerfile on this machine.
    Path(PathBuf),
    /// Literal Dockerfile text.
    Contents(String),
}

impl DockerfileInput {
    /// The Dockerfile text: literal contents as-is, a path by reading it.
    fn read(&self) -> Result<String, SailError> {
        let path = match self {
            DockerfileInput::Contents(text) => return Ok(text.clone()),
            DockerfileInput::Path(path) => path.as_path(),
        };
        // No path contains a newline; catching the one mistake this shape
        // invites here keeps a misplaced Dockerfile from surfacing as a
        // baffling missing-file error.
        if path.as_os_str().as_encoded_bytes().contains(&b'\n') {
            return Err(invalid(
                "the Dockerfile argument contains a newline, so it cannot be \
                 a path; pass literal Dockerfile text as contents"
                    .to_string(),
            ));
        }
        std::fs::read_to_string(path)
            .map_err(|err| invalid(format!("cannot read Dockerfile {}: {err}", path.display())))
    }

    /// The filesystem path this input names, when it names one.
    fn path(&self) -> Option<&Path> {
        match self {
            DockerfileInput::Path(path) => Some(path),
            DockerfileInput::Contents(_) => None,
        }
    }
}

/// A Dockerfile to build into an image, plus its build context. See
/// [`ImageDefinition::dockerfile`].
#[derive(Debug, Clone)]
pub struct DockerfileSource {
    /// The Dockerfile itself.
    pub dockerfile: DockerfileInput,
    /// Directory the Dockerfile's `COPY` and `ADD` instructions read from;
    /// `None` builds without a context.
    pub context_dir: Option<PathBuf>,
    /// Values for the Dockerfile's `ARG` instructions, like `--build-arg`.
    /// Names may not start with the reserved `BUILDKIT_` prefix, and Docker's
    /// proxy names (`HTTP_PROXY`, `HTTPS_PROXY`, `FTP_PROXY`, `NO_PROXY`,
    /// `ALL_PROXY`, in any letter case) are rejected; a step that needs a
    /// proxy can set one inside its `RUN` command.
    pub build_args: HashMap<String, String>,
    /// `.dockerignore`-style patterns excluding files from the context,
    /// applied after the context's own ignore file (a
    /// `<Dockerfile-name>.dockerignore` next to the Dockerfile when one
    /// exists, otherwise the context directory's `.dockerignore`) so they
    /// take precedence on conflict.
    pub ignore: Vec<String>,
}

/// 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` and
    /// `dockerfile`.
    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). A tag is pinned for your organization once an image
    /// has been built from it: later builds keep getting that version, even
    /// if the tag moves upstream. [`BuildMode::ForceBuild`] looks
    /// the tag up again and moves the pin for your whole organization. If
    /// forced builds of the same tag overlap, the last-requested one
    /// that succeeds decides what the tag means, no matter which build
    /// finishes first. A
    /// digest names exactly one image, so it never moves. The image's `ENV`,
    /// `WORKDIR`, and `USER` 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` and
    /// `dockerfile`.
    pub oci_ref: Option<String>,
    /// Your own Dockerfile built into the image. Every image its `FROM` (and
    /// `COPY --from`) instructions name must live on a supported public
    /// registry (`docker.io`, `ghcr.io`, `public.ecr.aws`, or `quay.io`; a
    /// short name like `python:3.12` means
    /// `docker.io/library/python:3.12`). Each named image is pinned to the
    /// version its tag pointed at the first time your organization used it, and
    /// those pinned versions become part of the built image's identity, so
    /// rebuilding the same definition reuses the same image even after a tag
    /// moves; a forced build looks the tags up again. A `# syntax=` line
    /// can declare `docker/dockerfile:1` or a release from 1.4 through
    /// 1.22.0; a file that declares anything else is rejected, and the
    /// declared release does not change how the file is built.
    /// Multi-stage Dockerfiles work. A `RUN --mount` of
    /// type `cache`, `secret`, or `ssh` is rejected; `tmpfs` mounts work, and
    /// `bind` mounts work when they read from the build context or another
    /// build stage. Mount options must be literal text, and `ONBUILD` is not
    /// supported, in the Dockerfile or in an image a `FROM` names. The built
    /// image's `ENV`, `WORKDIR`, and `USER` 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` and
    /// `oci_ref`.
    pub dockerfile: Option<DockerfileSource>,
    /// Target CPU architecture. Unspecified means amd64 with `base` and
    /// `dockerfile`, 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` or `dockerfile`: a pinned
    /// interpreter would shadow the Python the image was built around.
    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 or Dockerfile source always builds, so it is never builtin.
#[doc(hidden)]
pub fn is_builtin_base_spec(spec: &ImageSpec) -> bool {
    matches!(spec.base, Some(BaseImage::Debian | BaseImage::Devbox))
        && spec.oci.is_none()
        && spec.dockerfile.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 more than one
/// of the builtin base, OCI reference, and Dockerfile arms is ambiguous: the
/// build path would send one arm while `is_builtin_base_spec` classifies from
/// another, and Sailbox creation would serialize several and be rejected by
/// the backend as a duplicate oneof member. A directly constructed spec can
/// also carry unvalidated arm contents, so bound them here too. A spec with
/// no arm is the default image and is valid.
pub(crate) fn validate_image_spec_source(spec: &ImageSpec) -> Result<(), SailError> {
    let arms = usize::from(spec.base.is_some())
        + usize::from(spec.oci.is_some())
        + usize::from(spec.dockerfile.is_some());
    if arms > 1 {
        return Err(invalid(
            "an image takes one source: a builtin base, an OCI reference, or a Dockerfile"
                .to_string(),
        ));
    }
    if let Some(oci) = &spec.oci {
        if !spec.python_version.trim().is_empty() {
            return Err(invalid(
                "pythonVersion is not supported with an OCI reference: a pinned interpreter would shadow the Python the imported image was built around".to_string(),
            ));
        }
        validate_oci_ref(&oci.reference)?;
    }
    if let Some(dockerfile) = &spec.dockerfile {
        if !spec.python_version.trim().is_empty() {
            return Err(invalid(
                "pythonVersion is not supported with a Dockerfile: a pinned interpreter would shadow the Python the image was built around".to_string(),
            ));
        }
        validate_dockerfile_image(dockerfile)?;
    }
    Ok(())
}

/// Bound a Dockerfile source arm the way the backend does, so the common
/// mistakes fail before any hashing, uploading, or queueing. The service
/// parses the Dockerfile itself and stays authoritative on its contents.
fn validate_dockerfile_image(dockerfile: &crate::image::DockerfileImage) -> Result<(), SailError> {
    validate_dockerfile_text(&dockerfile.dockerfile)?;
    let entries = dockerfile.context_files.len()
        + dockerfile.context_dirs.len()
        + dockerfile.context_symlinks.len();
    if entries > MAX_DOCKERFILE_CONTEXT_FILES {
        return Err(invalid(format!(
            "dockerfile context has {entries} entries, max {MAX_DOCKERFILE_CONTEXT_FILES}"
        )));
    }
    validate_dockerfile_build_args(&dockerfile.build_args)
}

fn validate_dockerfile_text(text: &str) -> Result<(), SailError> {
    if text.trim().is_empty() {
        return Err(invalid("dockerfile text is required".to_string()));
    }
    if text.len() > MAX_DOCKERFILE_BYTES {
        return Err(invalid(format!(
            "dockerfile is {} bytes, max {MAX_DOCKERFILE_BYTES}",
            text.len()
        )));
    }
    Ok(())
}

/// The build arg names Docker reads, in any letter case, as proxy
/// configuration; the service rejects them (they would apply to build
/// steps without entering Docker's build cache keys).
const DOCKER_PROXY_BUILD_ARG_NAMES: [&str; 5] = [
    "http_proxy",
    "https_proxy",
    "ftp_proxy",
    "no_proxy",
    "all_proxy",
];

// Mirrors the service's build-arg rules so the common mistakes fail locally
// with the same boundaries: key count, shell-identifier keys within a byte
// cap, the reserved BUILDKIT_ prefix, Docker's proxy names, and value bytes
// free of characters that cannot survive a rendered build line.
fn validate_dockerfile_build_args(build_args: &HashMap<String, String>) -> Result<(), SailError> {
    if build_args.len() > MAX_DOCKERFILE_BUILD_ARGS {
        return Err(invalid(format!(
            "buildArgs has {} entries, max {MAX_DOCKERFILE_BUILD_ARGS}",
            build_args.len()
        )));
    }
    for (key, value) in build_args {
        if key.trim().is_empty() {
            return Err(invalid("buildArgs keys must be non-empty".to_string()));
        }
        if key.len() > MAX_DOCKERFILE_BUILD_ARG_KEY_BYTES || !is_shell_identifier(key) {
            return Err(invalid(format!(
                "buildArgs key {key:?} must match shell identifier syntax \
                 [A-Za-z_][A-Za-z0-9_]* within {MAX_DOCKERFILE_BUILD_ARG_KEY_BYTES} bytes"
            )));
        }
        if key.starts_with("BUILDKIT_") {
            return Err(invalid(format!(
                "buildArgs key {key:?} is reserved for the build system"
            )));
        }
        if DOCKER_PROXY_BUILD_ARG_NAMES
            .iter()
            .any(|name| key.eq_ignore_ascii_case(name))
        {
            return Err(invalid(format!(
                "buildArgs key {key:?} is a Docker proxy setting, which is not supported; \
                 set a proxy inside the RUN command that needs it"
            )));
        }
        if value.len() > MAX_DOCKERFILE_BUILD_ARG_VALUE_BYTES {
            return Err(invalid(format!(
                "buildArgs value for {key:?} is {} bytes, max {MAX_DOCKERFILE_BUILD_ARG_VALUE_BYTES}",
                value.len()
            )));
        }
        if value.contains(['\n', '\r', '\0']) {
            return Err(invalid(format!(
                "buildArgs value for {key:?} must not contain control characters"
            )));
        }
    }
    Ok(())
}

fn is_shell_identifier(key: &str) -> bool {
    let mut chars = key.chars();
    match chars.next() {
        Some(c) if c.is_ascii_alphabetic() || c == '_' => {}
        _ => return false,
    }
    chars.all(|c| c.is_ascii_alphanumeric() || c == '_')
}

/// 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();
    }
}

/// Carry a Dockerfile build's reference resolutions onto the spec, so
/// whatever is created from the spec names exactly the built bytes even
/// after a forced build moves what the references mean for the
/// organization. Specs with other sources, and responses without pins, are
/// left unchanged.
pub(crate) fn pin_dockerfile_from(spec: &mut ImageSpec, pins: Option<&[DockerfileFromResolution]>) {
    if let (Some(dockerfile), Some(pins)) = (spec.dockerfile.as_mut(), pins) {
        dockerfile.pinned_from = pins.to_vec();
    }
}

/// Decode a build response's Dockerfile pins; `None` on the wire means the
/// spec's source is not a Dockerfile.
fn dockerfile_pins_from_pb(
    pins: Option<pbimg::DockerfilePins>,
) -> Option<Vec<DockerfileFromResolution>> {
    pins.map(|pins| {
        pins.from_resolutions
            .into_iter()
            .map(|pin| DockerfileFromResolution {
                reference: pin.reference,
                digest_ref: pin.digest_ref,
            })
            .collect()
    })
}

/// 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}"),
    })?
}

#[derive(Debug)]
struct WalkedFile {
    abs_path: PathBuf,
    relative_path: String,
    mode: u32,
}

/// Everything a directory walk keeps. `addLocalDir` walks collect files
/// only; Dockerfile context walks also record directories and symbolic
/// links, because a docker build context carries both and `COPY` can name
/// them.
#[derive(Debug, Default)]
struct WalkedTree {
    files: Vec<WalkedFile>,
    dirs: Vec<crate::image::DockerfileContextDir>,
    symlinks: Vec<crate::image::DockerfileContextSymlink>,
}

impl WalkedTree {
    fn entries(&self) -> usize {
        self.files.len() + self.dirs.len() + self.symlinks.len()
    }
}

/// A [`WalkedTree`] after hashing and uploading: the content manifest plus
/// the walk's directories and symbolic links, each list path-sorted.
#[derive(Debug, Default)]
struct ResolvedDirTree {
    files: Vec<AddLocalDirFile>,
    dirs: Vec<crate::image::DockerfileContextDir>,
    symlinks: Vec<crate::image::DockerfileContextSymlink>,
}

/// The matcher a directory walk excludes files with.
enum WalkIgnore<'a> {
    /// Gitignore semantics, `addLocalDir`'s documented matching.
    Git(&'a ignore::gitignore::Gitignore),
    /// Docker's `.dockerignore` semantics, for Dockerfile build contexts.
    Docker(&'a crate::dockerignore::DockerPatternMatcher),
}

impl WalkIgnore<'_> {
    fn is_ignored(&self, rel_path: &str, is_dir: bool) -> Result<bool, SailError> {
        match self {
            WalkIgnore::Git(matcher) => Ok(matcher
                .matched_path_or_any_parents(rel_path, is_dir)
                .is_ignore()),
            WalkIgnore::Docker(matcher) => matcher.matches(rel_path).map_err(invalid),
        }
    }

    /// Whether the walk must still descend into an ignored directory.
    /// Gitignore cannot re-include below an excluded directory; Docker's
    /// `!` patterns can.
    fn descends_into_ignored_dirs(&self) -> bool {
        match self {
            WalkIgnore::Git(_) => false,
            WalkIgnore::Docker(matcher) => matcher.has_exclusions(),
        }
    }

    /// Whether the walk records directories and symbolic links alongside
    /// files. `addLocalDir` bakes files only (symlinks are documented as
    /// skipped); a Dockerfile context carries all three, like the context a
    /// docker build sends.
    fn records_dirs_and_symlinks(&self) -> bool {
        match self {
            WalkIgnore::Git(_) => false,
            WalkIgnore::Docker(_) => true,
        }
    }
}

/// Walk a local directory depth-first in sorted order, applying the given
/// ignore matching and enforcing the per-directory bounds. `op` names the
/// calling surface in error messages; `max_files` is that surface's cap on
/// kept entries.
fn walk_dir(
    root: &Path,
    matcher: &WalkIgnore<'_>,
    op: &str,
    max_files: usize,
) -> Result<WalkedTree, SailError> {
    fn check_cap(root: &Path, out: &WalkedTree, max_files: usize) -> Result<(), SailError> {
        if out.entries() > max_files {
            return Err(invalid(format!(
                "{} has more than {max_files} entries (max {max_files})",
                root.display()
            )));
        }
        Ok(())
    }

    fn check_rel_path(rel_path: &str) -> Result<(), SailError> {
        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"
            )));
        }
        Ok(())
    }

    fn recurse(
        root: &Path,
        dir: &Path,
        rel: &str,
        matcher: &WalkIgnore<'_>,
        op: &str,
        max_files: usize,
        out: &mut WalkedTree,
    ) -> 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!(
                        "{op}: {} 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() {
                if !matcher.records_dirs_and_symlinks()
                    || matcher.is_ignored(&rel_path, /* is_dir */ false)?
                {
                    continue;
                }
                check_rel_path(&rel_path)?;
                let target = std::fs::read_link(entry.path()).map_err(|err| {
                    invalid(format!(
                        "cannot read link {}: {err}",
                        entry.path().display()
                    ))
                })?;
                let target = target
                    .to_str()
                    .ok_or_else(|| {
                        invalid(format!(
                            "{op}: {} has a non-UTF-8 link target",
                            entry.path().display()
                        ))
                    })?
                    .to_string();
                out.symlinks.push(crate::image::DockerfileContextSymlink {
                    relative_path: rel_path,
                    target,
                });
                check_cap(root, out, max_files)?;
                continue;
            }
            let is_dir = file_type.is_dir();
            if matcher.is_ignored(&rel_path, is_dir)? {
                if is_dir && matcher.descends_into_ignored_dirs() {
                    let kept_before = out.entries();
                    recurse(root, &entry.path(), &rel_path, matcher, op, max_files, out)?;
                    if out.entries() > kept_before {
                        // A `!` pattern re-included something below, so the
                        // staged context needs this directory on the way
                        // down, exactly as a docker build sends it.
                        check_rel_path(&rel_path)?;
                        let metadata = entry.metadata().map_err(|err| {
                            invalid(format!("cannot stat {}: {err}", entry.path().display()))
                        })?;
                        check_context_mode_bits(op, &entry.path(), &metadata)?;
                        out.dirs.push(crate::image::DockerfileContextDir {
                            relative_path: rel_path,
                            mode: unix_mode(&metadata),
                        });
                        check_cap(root, out, max_files)?;
                    }
                }
                continue;
            }
            if is_dir {
                if matcher.records_dirs_and_symlinks() {
                    check_rel_path(&rel_path)?;
                    let metadata = entry.metadata().map_err(|err| {
                        invalid(format!("cannot stat {}: {err}", entry.path().display()))
                    })?;
                    check_context_mode_bits(op, &entry.path(), &metadata)?;
                    out.dirs.push(crate::image::DockerfileContextDir {
                        relative_path: rel_path.clone(),
                        mode: unix_mode(&metadata),
                    });
                    check_cap(root, out, max_files)?;
                }
                recurse(root, &entry.path(), &rel_path, matcher, op, max_files, out)?;
                continue;
            }
            if !file_type.is_file() {
                // `addLocalDir` bakes regular files only, so anything else
                // stays silently out of that walk. A Dockerfile context must
                // match what a docker build ships: sockets are skipped there
                // too, but a named pipe or device node cannot be content-
                // hashed, so the walk fails instead of silently building a
                // context that diverges from the local one.
                if matcher.records_dirs_and_symlinks() {
                    check_context_file_type(op, &entry.path(), file_type)?;
                }
                continue;
            }
            check_rel_path(&rel_path)?;
            let metadata = entry
                .metadata()
                .map_err(|err| invalid(format!("cannot stat {}: {err}", entry.path().display())))?;
            if matcher.records_dirs_and_symlinks() {
                check_context_mode_bits(op, &entry.path(), &metadata)?;
            }
            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.files.push(WalkedFile {
                abs_path: entry.path(),
                relative_path: rel_path,
                mode: unix_mode(&metadata),
            });
            check_cap(root, out, max_files)?;
        }
        Ok(())
    }

    let mut out = WalkedTree::default();
    recurse(root, root, "", matcher, op, max_files, &mut out)?;
    Ok(out)
}

/// How a directory walk decides which files are excluded.
enum DirWalkRules {
    /// Gitignore semantics, for `addLocalDir`. An empty walk is an error:
    /// baking an empty directory into an image is almost always a mistake.
    Gitignore {
        ignore: Vec<String>,
        ignore_file: Option<PathBuf>,
    },
    /// Docker's `.dockerignore` semantics, for Dockerfile build contexts:
    /// the walk alone decides what ships (the build uses the shipped
    /// context as-is), so its verdicts must be the ones Docker's own
    /// matcher would reach. An empty walk is legal here; a COPY-less
    /// Dockerfile needs no context.
    DockerContext { patterns: Vec<String> },
}

/// The blocking half of [`Client::resolve_dir_files`]: check the root is a
/// directory, build the ignore matcher, walk, and apply the empty-walk
/// policy. Free of the client so the filesystem semantics are testable alone.
fn walk_dir_files(
    op: &str,
    root: &Path,
    rules: &DirWalkRules,
    max_files: usize,
) -> Result<WalkedTree, SailError> {
    let metadata = std::fs::metadata(root).map_err(|_| {
        invalid(format!(
            "{op}: {} does not exist or is not a directory",
            root.display()
        ))
    })?;
    if !metadata.is_dir() {
        return Err(invalid(format!(
            "{op}: {} is not a directory",
            root.display()
        )));
    }
    match rules {
        DirWalkRules::Gitignore {
            ignore,
            ignore_file,
        } => {
            let matcher = ignore_matcher(root, ignore, ignore_file.as_deref())?;
            let walked = walk_dir(root, &WalkIgnore::Git(&matcher), op, max_files)?;
            if walked.files.is_empty() {
                let qualifier = if !ignore.is_empty() || ignore_file.is_some() {
                    " after applying ignore patterns"
                } else {
                    ""
                };
                return Err(invalid(format!(
                    "{op}: {} contains no files{qualifier}",
                    root.display()
                )));
            }
            Ok(walked)
        }
        DirWalkRules::DockerContext { patterns } => {
            let matcher =
                crate::dockerignore::DockerPatternMatcher::new(patterns).map_err(invalid)?;
            walk_dir(root, &WalkIgnore::Docker(&matcher), op, max_files)
        }
    }
}

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

/// A Dockerfile context ships regular files, directories, and symbolic
/// links. Sockets are silently skipped, matching docker, whose context tar
/// cannot carry one. A named pipe or device node has no hashable content,
/// so it fails the walk. Hard links are the one silent divergence from a
/// local docker build: the manifest ships each path's bytes independently,
/// so two names for one file arrive in the build as two files.
#[cfg(unix)]
fn check_context_file_type(
    op: &str,
    path: &Path,
    file_type: std::fs::FileType,
) -> Result<(), SailError> {
    use std::os::unix::fs::FileTypeExt;
    if file_type.is_socket() {
        return Ok(());
    }
    Err(invalid(format!(
        "{op}: {} is a named pipe or device node; a build context can carry only regular files, directories, and symbolic links",
        path.display()
    )))
}

#[cfg(not(unix))]
fn check_context_file_type(
    _op: &str,
    _path: &Path,
    _file_type: std::fs::FileType,
) -> Result<(), SailError> {
    Ok(())
}

/// The context manifest records only the lower permission bits, so a
/// setuid, setgid, or sticky bit would be silently stripped. For
/// `addLocalDir` that stripping is shipped behavior, but a Dockerfile
/// context promises the build sees what a local docker build would, and a
/// COPY'd binary quietly losing its setuid bit breaks that, so the walk
/// refuses instead. Mode 000 breaks the same promise from the other end:
/// the manifest cannot tell an explicit 000 from an unrecorded mode, so
/// the build would substitute default permissions for it.
#[cfg(unix)]
fn check_context_mode_bits(
    op: &str,
    path: &Path,
    metadata: &std::fs::Metadata,
) -> Result<(), SailError> {
    use std::os::unix::fs::PermissionsExt;
    let mode = metadata.permissions().mode();
    if mode & 0o7000 != 0 {
        return Err(invalid(format!(
            "{op}: {} has a setuid, setgid, or sticky permission bit, which a build context does not preserve; clear the bit or exclude the path",
            path.display()
        )));
    }
    let permission_bits = mode & 0o777;
    if permission_bits == 0 {
        return Err(invalid(format!(
            "{op}: {} has no permission bits (mode 000), which a build context does not preserve; add a permission bit or exclude the path",
            path.display()
        )));
    }
    Ok(())
}

#[cfg(not(unix))]
fn check_context_mode_bits(
    _op: &str,
    _path: &Path,
    _metadata: &std::fs::Metadata,
) -> Result<(), SailError> {
    Ok(())
}

#[cfg(not(unix))]
fn unix_mode(metadata: &std::fs::Metadata) -> u32 {
    non_unix_mode(metadata.is_dir())
}

/// The recorded mode on platforms without Unix permission bits: regular files
/// read-write, directories additionally executable so they stay traversable
/// when the build runs as a non-root `USER`.
#[cfg(any(not(unix), test))]
fn non_unix_mode(is_dir: bool) -> u32 {
    if is_dir {
        0o755
    } else {
        0o644
    }
}

/// The `<Dockerfile-name>.dockerignore` path next to a Dockerfile, which
/// Docker prefers over the context directory's `.dockerignore` whenever
/// it exists.
fn sibling_dockerignore(dockerfile: &Path) -> PathBuf {
    let mut name = dockerfile.file_name().unwrap_or_default().to_os_string();
    name.push(".dockerignore");
    dockerfile.with_file_name(name)
}

/// The effective ignore text the context walk parses when explicit ignore
/// patterns extend an on-disk `.dockerignore`: the original bytes with each
/// pattern appended as its own line, so on conflict the appended patterns
/// win (the last matching pattern decides). Only the walk consumes this
/// text; the staged context receives no further ignore processing, and a
/// retained `.dockerignore` ships as the ordinary file it is.
fn extended_dockerignore(original: &[u8], patterns: &[String]) -> Vec<u8> {
    let mut extended = original.to_vec();
    if !extended.is_empty() && !extended.ends_with(b"\n") {
        extended.push(b'\n');
    }
    for pattern in patterns {
        extended.extend_from_slice(pattern.as_bytes());
        extended.push(b'\n');
    }
    extended
}

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(local_dir_file_to_pb).collect(),
        }),
    };
    pbimage::ImageBuildStep { step: Some(inner) }
}

fn local_dir_file_to_pb(file: &AddLocalDirFile) -> pbimage::AddLocalDirFile {
    pbimage::AddLocalDirFile {
        relative_path: file.relative_path.clone(),
        content_sha256: file.content_sha256.clone(),
        mode: file.mode,
    }
}

/// 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.dockerfile, spec.base) {
        (Some(oci), _, _) => Some(pbimage::image_spec::Source::Oci(pbimage::OciImage {
            r#ref: oci.reference.clone(),
        })),
        (None, Some(dockerfile), _) => Some(pbimage::image_spec::Source::Dockerfile(
            pbimage::DockerfileImage {
                dockerfile: dockerfile.dockerfile.clone(),
                context_files: dockerfile
                    .context_files
                    .iter()
                    .map(local_dir_file_to_pb)
                    .collect(),
                build_args: dockerfile.build_args.clone(),
                context_dirs: dockerfile
                    .context_dirs
                    .iter()
                    .map(|dir| pbimage::DockerfileContextDir {
                        relative_path: dir.relative_path.clone(),
                        mode: dir.mode,
                    })
                    .collect(),
                context_symlinks: dockerfile
                    .context_symlinks
                    .iter()
                    .map(|link| pbimage::DockerfileContextSymlink {
                        relative_path: link.relative_path.clone(),
                        target: link.target.clone(),
                    })
                    .collect(),
                pinned_from: dockerfile
                    .pinned_from
                    .iter()
                    .map(|pin| pbimage::DockerfileFromResolution {
                        reference: pin.reference.clone(),
                        digest_ref: pin.digest_ref.clone(),
                    })
                    .collect(),
            },
        )),
        (None, None, Some(base)) => Some(pbimage::image_spec::Source::Base(
            base_image_to_pb(base) as i32
        )),
        (None, 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,
            retryable: response.retryable,
            resolved_oci_ref: response.resolved_oci_ref,
            dockerfile_pins: dockerfile_pins_from_pb(response.dockerfile_pins),
        })
    }

    /// 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,
            retryable: response.retryable,
            resolved_oci_ref: response.resolved_oci_ref,
            dockerfile_pins: dockerfile_pins_from_pb(response.dockerfile_pins),
        })
    }

    /// 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)?;
        let resolved = self
            .resolve_dir_files(
                "addLocalDir",
                local_path,
                DirWalkRules::Gitignore {
                    ignore: ignore.to_vec(),
                    ignore_file: ignore_file.map(Path::to_path_buf),
                },
                MAX_LOCAL_DIR_FILES,
            )
            .await?;
        Ok(crate::image::AddLocalDir {
            remote_path: target,
            files: resolved.files,
        })
    }

    /// Walk a local directory with the given ignore rules, hash every file,
    /// and upload content the server does not already have, returning the
    /// path-sorted content manifest (plus the walk's directories and symbolic
    /// links, which carry no content). `op` names the calling surface in
    /// error messages.
    async fn resolve_dir_files(
        &self,
        op: &'static str,
        local_path: &Path,
        rules: DirWalkRules,
        max_files: usize,
    ) -> Result<ResolvedDirTree, SailError> {
        // 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 walked =
            tokio::task::spawn_blocking(move || walk_dir_files(op, &walk_root, &rules, max_files))
                .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.files.len());
        for file in walked.files {
            let (digest, size) = hash_file(&file.abs_path).await?;
            if size > MAX_LOCAL_FILE_BYTES {
                return Err(invalid(format!(
                    "{op}: {} ({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?;
        // Walk order is not path order: a re-included directory is pushed
        // after its descendants. Sort so the manifest is canonical.
        let mut dirs = walked.dirs;
        dirs.sort_by(|a, b| a.relative_path.cmp(&b.relative_path));
        let mut symlinks = walked.symlinks;
        symlinks.sort_by(|a, b| a.relative_path.cmp(&b.relative_path));
        Ok(ResolvedDirTree {
            files,
            dirs,
            symlinks,
        })
    }

    /// 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 definition that sets more
        // than one image source fails before those network side effects. The
        // probe carries the trimmed reference and the Dockerfile text (read
        // here, a cheap local step) so the validator sees the exact values
        // the final spec will carry.
        let oci = def.oci_ref.as_deref().map(|raw| OciImage {
            reference: raw.trim().to_string(),
        });
        let dockerfile_text = match &def.dockerfile {
            Some(source) => Some(source.dockerfile.read()?),
            None => None,
        };
        validate_image_spec_source(&ImageSpec {
            base: def.base,
            oci: oci.clone(),
            dockerfile: def.dockerfile.as_ref().zip(dockerfile_text.as_ref()).map(
                |(source, text)| crate::image::DockerfileImage {
                    dockerfile: text.clone(),
                    build_args: source.build_args.clone(),
                    ..Default::default()
                },
            ),
            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?,
                ),
            });
        }
        let dockerfile = match def.dockerfile.as_ref().zip(dockerfile_text) {
            Some((source, text)) => Some(self.resolve_dockerfile_context(source, text).await?),
            None => None,
        };
        Ok(ImageSpec {
            base: def.base,
            oci,
            dockerfile,
            build_steps: steps,
            env: def.env.clone(),
            architecture: def.architecture,
            python_version: def.python_version.clone(),
            filesystem: def.filesystem,
        })
    }

    /// Resolve a [`DockerfileSource`] into the wire's
    /// [`DockerfileImage`](crate::image::DockerfileImage): read the
    /// Dockerfile, then walk, hash, and upload its build context.
    #[doc(hidden)]
    pub async fn resolve_dockerfile_source(
        &self,
        source: &DockerfileSource,
    ) -> Result<crate::image::DockerfileImage, SailError> {
        let text = source.dockerfile.read()?;
        validate_dockerfile_text(&text)?;
        validate_dockerfile_build_args(&source.build_args)?;
        self.resolve_dockerfile_context(source, text).await
    }

    /// The context half of [`Client::resolve_dockerfile_source`], for a
    /// caller that already read and validated the Dockerfile text. The
    /// context's ignore file is honored with Docker's own matching rules
    /// and Docker's own selection: a `<Dockerfile-name>.dockerignore`
    /// next to a Dockerfile given as a path wins by existing — even
    /// empty — and only otherwise does the context directory's
    /// `.dockerignore` apply. The source's explicit ignore patterns are
    /// appended after the file's lines, so on conflict the explicit
    /// patterns win (the last matching pattern decides). The rules
    /// filter the walk only: the manifest is the final context — the
    /// build applies no further ignore rules to it — so a retained
    /// `.dockerignore` ships as the ordinary file or symlink it is,
    /// exactly the bytes a `COPY . /` puts in the image locally.
    async fn resolve_dockerfile_context(
        &self,
        source: &DockerfileSource,
        dockerfile: String,
    ) -> Result<crate::image::DockerfileImage, SailError> {
        let context = match &source.context_dir {
            Some(context_dir) => {
                let sibling = source.dockerfile.path().map(sibling_dockerignore);
                let dockerignore = match sibling {
                    Some(path) if path.is_file() => path,
                    _ => context_dir.join(".dockerignore"),
                };
                let original = if dockerignore.is_file() {
                    Some(tokio::fs::read(&dockerignore).await.map_err(|err| {
                        invalid(format!("cannot read {}: {err}", dockerignore.display()))
                    })?)
                } else {
                    None
                };
                let effective =
                    extended_dockerignore(original.as_deref().unwrap_or_default(), &source.ignore);
                let patterns = crate::dockerignore::read_patterns(&effective).map_err(invalid)?;
                self.resolve_dir_files(
                    "contextDir",
                    context_dir,
                    DirWalkRules::DockerContext { patterns },
                    MAX_DOCKERFILE_CONTEXT_FILES,
                )
                .await?
            }
            None => ResolvedDirTree::default(),
        };
        Ok(crate::image::DockerfileImage {
            dockerfile,
            context_files: context.files,
            build_args: source.build_args.clone(),
            context_dirs: context.dirs,
            context_symlinks: context.symlinks,
            // A freshly resolved definition has no build behind it yet;
            // pins arrive on the spec a completed build returns.
            pinned_from: Vec::new(),
        })
    }

    /// Best-effort refresh of a pinned Dockerfile context, bounded by
    /// `timeout`: for each manifest entry whose file under `context_dir`
    /// still hashes to its pinned digest, re-upload the content if the
    /// server no longer holds it (re-marking held content as recently
    /// used). An entry whose file drifted or disappeared is skipped: the
    /// pinned spec cannot use its current bytes, and nothing outside the
    /// manifest is ever read or uploaded.
    #[doc(hidden)]
    pub async fn refresh_dockerfile_context(
        &self,
        context_dir: &Path,
        files: &[AddLocalDirFile],
        timeout: Duration,
    ) -> Result<(), SailError> {
        tokio::time::timeout(timeout, async {
            // digest -> (source path, size); deduped so shared content
            // uploads once.
            let mut uploads: HashMap<String, (PathBuf, u64)> = HashMap::new();
            for file in files {
                if uploads.contains_key(&file.content_sha256) {
                    continue;
                }
                let path = context_dir.join(&file.relative_path);
                let Ok((digest, size)) = hash_file(&path).await else {
                    continue;
                };
                if digest != file.content_sha256 {
                    continue;
                }
                uploads.insert(digest, (path, size));
            }
            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
        })
        .await
        .map_err(|_| SailError::Transport {
            kind: TransportKind::Timeout,
            message: "pinned Dockerfile context refresh did not finish in time".to_string(),
            source: None,
        })?
    }

    /// 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, BuildOrigin::DirectRequest, mode)
                    .await
            }
            Some(_) => tokio::time::timeout(
                timeout,
                self.build_spec_ready_cached(spec, timeout, BuildOrigin::DirectRequest, 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,
        origin: BuildOrigin,
        mode: BuildMode,
    ) -> Result<ImageBuild, SailError> {
        let key = canonical_spec_key(spec)?;
        let retain_ready = crate::imagecache::retains_ready(spec);
        loop {
            let joined = self
                .image_ready_cache()
                .join_or_lead(&key, origin, mode, |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
    /// base image skips the build. `timeout` bounds the whole pipeline,
    /// including hashing, uploads, the build, and any automatic retries;
    /// 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, BuildOrigin::DirectRequest, 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);
            pin_dockerfile_from(&mut spec, build.dockerfile_pins.as_deref());
            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 retry_spec = spec.clone();
        let mut build = self.build_image(&retry_spec, rpc_budget(), mode).await?;
        loop {
            match build.status {
                ImageBuildStatus::Ready => return Ok(build),
                ImageBuildStatus::Failed => {
                    if build.retryable || build.error_message == GUEST_SCHEMA_SUPERSEDED_MESSAGE {
                        // A guest-schema change creates a new immutable image
                        // identity. Reuse builds keep the registry bytes chosen
                        // by the interrupted build. Force builds deliberately
                        // keep their original spec and mode: for a mutable tag,
                        // the service must record that tag on the replacement
                        // attempt so readiness can publish the new digest back
                        // to it.
                        if mode == BuildMode::ReuseExisting {
                            pin_resolved_oci_ref(&mut retry_spec, &build.resolved_oci_ref);
                            pin_dockerfile_from(&mut retry_spec, build.dockerfile_pins.as_deref());
                        }
                        tokio::time::sleep(next_build_poll_delay(deadline, &build.image_id)?).await;
                        build = self.build_image(&retry_spec, rpc_budget(), mode).await?;
                        continue;
                    }
                    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 = next_build_poll_delay(deadline, &build.image_id)?;
            tokio::time::sleep(nap).await;
            build = self
                .get_image_build_status(&build.image_id, rpc_budget())
                .await?;
        }
    }
}

fn next_build_poll_delay(deadline: Option<Instant>, image_id: &str) -> Result<Duration, SailError> {
    let Some(deadline) = deadline else {
        return Ok(BUILD_POLL_INTERVAL);
    };
    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 {image_id}"),
            source: None,
        });
    }
    Ok(left.min(BUILD_POLL_INTERVAL))
}

/// 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 the_non_unix_mode_fallback_keeps_directories_traversable() {
        assert_eq!(super::non_unix_mode(false), 0o644);
        assert_eq!(super::non_unix_mode(true), 0o755);
    }

    #[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 (a pinned
        // interpreter would shadow the image's own Python); 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(),
            &WalkIgnore::Git(&matcher),
            "addLocalDir",
            MAX_LOCAL_DIR_FILES,
        )
        .expect("walk");
        let mut paths: Vec<_> = walked
            .files
            .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"
        );
    }

    #[test]
    fn dockerfile_input_reads_contents_and_paths() {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("Dockerfile");
        std::fs::write(&path, "FROM python:3.12\n").unwrap();

        // Literal contents pass through untouched, newline or not.
        let contents = DockerfileInput::Contents("FROM scratch".to_string());
        assert_eq!(contents.read().unwrap(), "FROM scratch");
        let multiline = DockerfileInput::Contents("FROM scratch\nRUN true".to_string());
        assert_eq!(multiline.read().unwrap(), "FROM scratch\nRUN true");
        let by_path = DockerfileInput::Path(path);
        assert_eq!(by_path.read().unwrap(), "FROM python:3.12\n");

        // Dockerfile text passed where a path belongs fails on the newline
        // itself, before any filesystem probe turns it into a missing-file
        // error.
        let mixup = DockerfileInput::Path(PathBuf::from("FROM scratch\nRUN true"))
            .read()
            .unwrap_err()
            .to_string();
        assert!(mixup.contains("cannot be a path"), "{mixup}");
        assert!(mixup.contains("as contents"), "{mixup}");
        let missing = dir.path().join("absent");
        let path_err = DockerfileInput::Path(missing)
            .read()
            .unwrap_err()
            .to_string();
        assert!(path_err.contains("cannot read Dockerfile"), "{path_err}");
        assert!(!path_err.contains("cannot be a path"), "{path_err}");
    }

    #[test]
    fn dockerfile_spec_validation_matches_the_backend_bounds() {
        use crate::image::{DockerfileImage, ImageSpec, OciImage};
        let dockerfile_spec = |text: &str| ImageSpec {
            dockerfile: Some(DockerfileImage {
                dockerfile: text.to_string(),
                ..Default::default()
            }),
            ..Default::default()
        };
        // A well-formed dockerfile-only spec passes and always builds.
        let good = dockerfile_spec("FROM python:3.12\nRUN true");
        assert!(validate_image_spec_source(&good).is_ok());
        assert!(!is_builtin_base_spec(&good));
        // More than one source arm is ambiguous.
        let with_base = ImageSpec {
            base: Some(BaseImage::Debian),
            ..good.clone()
        };
        assert!(!is_builtin_base_spec(&with_base));
        assert!(validate_image_spec_source(&with_base).is_err());
        let with_oci = ImageSpec {
            oci: Some(OciImage {
                reference: format!("docker.io/library/ubuntu@sha256:{}", "a".repeat(64)),
            }),
            ..good.clone()
        };
        assert!(validate_image_spec_source(&with_oci).is_err());
        // A pinned interpreter would shadow the Python the image was
        // built around.
        let with_python = ImageSpec {
            python_version: "3.12.13".to_string(),
            ..good.clone()
        };
        assert!(validate_image_spec_source(&with_python).is_err());
        // Text bounds: blank and oversized fail; exactly the cap passes.
        assert!(validate_image_spec_source(&dockerfile_spec("  \n ")).is_err());
        assert!(
            validate_image_spec_source(&dockerfile_spec(&"x".repeat(MAX_DOCKERFILE_BYTES))).is_ok()
        );
        assert!(validate_image_spec_source(&dockerfile_spec(
            &"x".repeat(MAX_DOCKERFILE_BYTES + 1)
        ))
        .is_err());
        // The context manifest is capped.
        let mut crowded = good.clone();
        crowded.dockerfile.as_mut().unwrap().context_files =
            vec![AddLocalDirFile::default(); MAX_DOCKERFILE_CONTEXT_FILES + 1];
        assert!(validate_image_spec_source(&crowded).is_err());
        // Build-arg keys must be non-empty.
        let mut blank_key = good.clone();
        blank_key
            .dockerfile
            .as_mut()
            .unwrap()
            .build_args
            .insert(" ".to_string(), "value".to_string());
        assert!(validate_image_spec_source(&blank_key).is_err());
        // Build args follow the service's bounds exactly: count, key shape
        // and byte length, the reserved BUILDKIT_ prefix, Docker's proxy
        // names, and value bytes free of newlines and NUL.
        let with_args = |args: &[(&str, &str)]| {
            let mut spec = good.clone();
            spec.dockerfile.as_mut().unwrap().build_args = args
                .iter()
                .map(|(k, v)| (k.to_string(), v.to_string()))
                .collect();
            spec
        };
        let max_args: Vec<(String, String)> = (0..MAX_DOCKERFILE_BUILD_ARGS)
            .map(|i| (format!("ARG_{i}"), "v".to_string()))
            .collect();
        let max_refs: Vec<(&str, &str)> = max_args
            .iter()
            .map(|(k, v)| (k.as_str(), v.as_str()))
            .collect();
        assert!(validate_image_spec_source(&with_args(&max_refs)).is_ok());
        let mut over = with_args(&max_refs);
        over.dockerfile
            .as_mut()
            .unwrap()
            .build_args
            .insert("ONE_MORE".to_string(), "v".to_string());
        assert!(validate_image_spec_source(&over).is_err());
        assert!(validate_image_spec_source(&with_args(&[("1BAD", "v")])).is_err());
        assert!(validate_image_spec_source(&with_args(&[("WITH-DASH", "v")])).is_err());
        let long_key = "K".repeat(MAX_DOCKERFILE_BUILD_ARG_KEY_BYTES + 1);
        assert!(validate_image_spec_source(&with_args(&[(&long_key, "v")])).is_err());
        assert!(validate_image_spec_source(&with_args(&[("BUILDKIT_SYNTAX", "v")])).is_err());
        // Docker's proxy names are rejected as the frontend matches them:
        // whole-key and case-insensitive.
        assert!(validate_image_spec_source(&with_args(&[("HTTP_PROXY", "v")])).is_err());
        assert!(validate_image_spec_source(&with_args(&[("https_proxy", "v")])).is_err());
        assert!(validate_image_spec_source(&with_args(&[("All_Proxy", "v")])).is_err());
        assert!(validate_image_spec_source(&with_args(&[("MY_HTTP_PROXY", "v")])).is_ok());
        // Value bytes, not characters: 3000 two-byte characters exceed the cap.
        let multibyte = "é".repeat(3000);
        assert!(validate_image_spec_source(&with_args(&[("KEY", &multibyte)])).is_err());
        let max_value = "v".repeat(MAX_DOCKERFILE_BUILD_ARG_VALUE_BYTES);
        assert!(validate_image_spec_source(&with_args(&[("KEY", &max_value)])).is_ok());
        assert!(validate_image_spec_source(&with_args(&[("KEY", "a\nb")])).is_err());
        assert!(validate_image_spec_source(&with_args(&[("KEY", "a\rb")])).is_err());
        assert!(validate_image_spec_source(&with_args(&[("KEY", "a\0b")])).is_err());
    }

    #[test]
    fn dockerfile_spec_maps_to_the_dockerfile_oneof_arm() {
        use crate::image::{
            DockerfileContextDir, DockerfileContextSymlink, DockerfileImage, ImageSpec,
        };
        let spec = ImageSpec {
            dockerfile: Some(DockerfileImage {
                dockerfile: "FROM python:3.12".to_string(),
                context_files: vec![AddLocalDirFile {
                    relative_path: "app/main.py".to_string(),
                    content_sha256: "a".repeat(64),
                    mode: 0o755,
                }],
                build_args: HashMap::from([("VERSION".to_string(), "1".to_string())]),
                context_dirs: vec![DockerfileContextDir {
                    relative_path: "empty".to_string(),
                    mode: 0o700,
                }],
                context_symlinks: vec![DockerfileContextSymlink {
                    relative_path: "link.py".to_string(),
                    target: "app/main.py".to_string(),
                }],
                pinned_from: vec![crate::image::DockerfileFromResolution {
                    reference: "docker.io/library/python:3.12".to_string(),
                    digest_ref: format!("docker.io/library/python@sha256:{}", "a".repeat(64)),
                }],
            }),
            ..Default::default()
        };
        let pb = image_spec_to_pb(&spec);
        match pb.source {
            Some(crate::pb::image::v1::image_spec::Source::Dockerfile(dockerfile)) => {
                assert_eq!(dockerfile.dockerfile, "FROM python:3.12");
                assert_eq!(dockerfile.context_files.len(), 1);
                assert_eq!(dockerfile.context_files[0].relative_path, "app/main.py");
                assert_eq!(dockerfile.context_files[0].mode, 0o755);
                assert_eq!(dockerfile.build_args["VERSION"], "1");
                assert_eq!(dockerfile.context_dirs.len(), 1);
                assert_eq!(dockerfile.context_dirs[0].relative_path, "empty");
                assert_eq!(dockerfile.context_dirs[0].mode, 0o700);
                assert_eq!(dockerfile.context_symlinks.len(), 1);
                assert_eq!(dockerfile.context_symlinks[0].relative_path, "link.py");
                assert_eq!(dockerfile.context_symlinks[0].target, "app/main.py");
                assert_eq!(dockerfile.pinned_from.len(), 1);
                assert_eq!(
                    dockerfile.pinned_from[0].reference,
                    "docker.io/library/python:3.12"
                );
                assert_eq!(
                    dockerfile.pinned_from[0].digest_ref,
                    format!("docker.io/library/python@sha256:{}", "a".repeat(64))
                );
            }
            other => panic!("pb source = {other:?}, want the dockerfile arm"),
        }
    }

    // A completed build's pins land on the spec's dockerfile arm and only
    // there: other sources and pin-less responses leave the spec unchanged.
    #[test]
    fn pin_dockerfile_from_carries_pins_onto_the_dockerfile_arm() {
        use crate::image::{DockerfileFromResolution, DockerfileImage, ImageSpec, OciImage};
        let pins = vec![DockerfileFromResolution {
            reference: "docker.io/library/python:3.12".to_string(),
            digest_ref: format!("docker.io/library/python@sha256:{}", "a".repeat(64)),
        }];
        let mut spec = ImageSpec {
            dockerfile: Some(DockerfileImage {
                dockerfile: "FROM python:3.12".to_string(),
                ..Default::default()
            }),
            ..Default::default()
        };
        pin_dockerfile_from(&mut spec, None);
        assert!(spec.dockerfile.as_ref().unwrap().pinned_from.is_empty());
        pin_dockerfile_from(&mut spec, Some(&pins));
        assert_eq!(spec.dockerfile.as_ref().unwrap().pinned_from, pins);
        let mut oci = ImageSpec {
            oci: Some(OciImage {
                reference: "docker.io/library/ubuntu:24.04".to_string(),
            }),
            ..Default::default()
        };
        pin_dockerfile_from(&mut oci, Some(&pins));
        assert!(oci.dockerfile.is_none());
    }

    fn docker_context_rules(dockerignore: &[u8], ignore: &[&str]) -> DirWalkRules {
        let ignore: Vec<String> = ignore.iter().map(ToString::to_string).collect();
        DirWalkRules::DockerContext {
            patterns: crate::dockerignore::read_patterns(&extended_dockerignore(
                dockerignore,
                &ignore,
            ))
            .expect("test patterns fit the line bound"),
        }
    }

    #[test]
    fn dockerfile_context_walk_allows_empty_and_caps_file_count() {
        let dir = tempfile::tempdir().expect("tempdir");
        // An empty walk is an error for addLocalDir but legal for a
        // Dockerfile build context (a COPY-less Dockerfile is fine).
        let err = walk_dir_files(
            "addLocalDir",
            dir.path(),
            &DirWalkRules::Gitignore {
                ignore: Vec::new(),
                ignore_file: None,
            },
            MAX_LOCAL_DIR_FILES,
        )
        .unwrap_err()
        .to_string();
        assert!(err.contains("contains no files"), "{err}");
        let empty = walk_dir_files(
            "contextDir",
            dir.path(),
            &docker_context_rules(b"", &[]),
            MAX_DOCKERFILE_CONTEXT_FILES,
        )
        .expect("empty context");
        assert_eq!(empty.entries(), 0);

        // The count cap fails during the walk, before any hashing.
        for i in 0..3 {
            std::fs::write(dir.path().join(format!("file{i}")), b"x").unwrap();
        }
        let err = walk_dir_files("contextDir", dir.path(), &docker_context_rules(b"", &[]), 2)
            .unwrap_err()
            .to_string();
        assert!(err.contains("more than 2 entries"), "{err}");
    }

    #[test]
    fn explicit_ignore_patterns_override_the_dockerignore_file() {
        let dir = tempfile::tempdir().expect("tempdir");
        std::fs::write(dir.path().join("keep.log"), b"keep").unwrap();
        std::fs::write(dir.path().join("drop.log"), b"drop").unwrap();
        std::fs::write(dir.path().join("app.py"), b"app").unwrap();

        // The ignore file's patterns apply; explicit patterns come after its
        // lines, so a `!` re-include wins (the last matching pattern
        // decides), mirroring how resolve_dockerfile_context layers them.
        let walked = walk_dir_files(
            "contextDir",
            dir.path(),
            &docker_context_rules(b"*.log\n", &["!keep.log"]),
            MAX_DOCKERFILE_CONTEXT_FILES,
        )
        .expect("walk");
        let mut paths: Vec<_> = walked
            .files
            .iter()
            .map(|f| f.relative_path.clone())
            .collect();
        paths.sort();
        assert_eq!(paths, ["app.py", "keep.log"]);
    }

    #[test]
    fn dockerfile_context_walk_uses_docker_ignore_semantics() {
        // `[!a].txt` under gitignore is negation ("anything but a"); under
        // Docker it is a literal class containing `!` and `a`. The walk must
        // read it Docker's way: a.txt excluded, b.txt kept.
        let dir = tempfile::tempdir().expect("tempdir");
        std::fs::write(dir.path().join("a.txt"), b"a").unwrap();
        std::fs::write(dir.path().join("b.txt"), b"b").unwrap();
        let walked = walk_dir_files(
            "contextDir",
            dir.path(),
            &docker_context_rules(b"[!a].txt\n", &[]),
            MAX_DOCKERFILE_CONTEXT_FILES,
        )
        .expect("walk");
        let paths: Vec<_> = walked
            .files
            .iter()
            .map(|f| f.relative_path.clone())
            .collect();
        assert_eq!(paths, ["b.txt"]);
    }

    #[test]
    fn dockerfile_context_walk_reincludes_under_an_excluded_directory() {
        // Docker's `!` patterns reach below an excluded directory, so the
        // walk must descend into it; gitignore matching would prune the
        // directory and never see keep.log.
        let dir = tempfile::tempdir().expect("tempdir");
        std::fs::create_dir(dir.path().join("logs")).unwrap();
        std::fs::write(dir.path().join("logs/keep.log"), b"keep").unwrap();
        std::fs::write(dir.path().join("logs/drop.log"), b"drop").unwrap();
        let walked = walk_dir_files(
            "contextDir",
            dir.path(),
            &docker_context_rules(b"logs\n!logs/keep.log\n", &[]),
            MAX_DOCKERFILE_CONTEXT_FILES,
        )
        .expect("walk");
        let paths: Vec<_> = walked
            .files
            .iter()
            .map(|f| f.relative_path.clone())
            .collect();
        assert_eq!(paths, ["logs/keep.log"]);
        // The excluded directory itself comes along because something under
        // it was kept: the staged context needs it on the way down.
        let dirs: Vec<_> = walked
            .dirs
            .iter()
            .map(|d| d.relative_path.clone())
            .collect();
        assert_eq!(dirs, ["logs"]);
    }

    #[test]
    fn dockerfile_context_walk_records_dirs_and_symlinks() {
        let dir = tempfile::tempdir().expect("tempdir");
        std::fs::create_dir(dir.path().join("empty")).unwrap();
        std::fs::create_dir(dir.path().join("sub")).unwrap();
        std::fs::write(dir.path().join("sub/app.py"), b"app").unwrap();
        std::os::unix::fs::symlink("sub/app.py", dir.path().join("link.py")).unwrap();
        std::os::unix::fs::symlink("/etc/hosts", dir.path().join("abs.link")).unwrap();
        let walked = walk_dir_files(
            "contextDir",
            dir.path(),
            &docker_context_rules(b"", &[]),
            MAX_DOCKERFILE_CONTEXT_FILES,
        )
        .expect("walk");
        let files: Vec<_> = walked
            .files
            .iter()
            .map(|f| f.relative_path.clone())
            .collect();
        assert_eq!(files, ["sub/app.py"]);
        let dirs: Vec<_> = walked
            .dirs
            .iter()
            .map(|d| d.relative_path.clone())
            .collect();
        assert_eq!(dirs, ["empty", "sub"]);
        let links: Vec<_> = walked
            .symlinks
            .iter()
            .map(|s| (s.relative_path.clone(), s.target.clone()))
            .collect();
        assert_eq!(
            links,
            [
                ("abs.link".to_string(), "/etc/hosts".to_string()),
                ("link.py".to_string(), "sub/app.py".to_string()),
            ]
        );
    }

    #[test]
    fn dockerfile_context_walk_ignores_dirs_and_symlinks_by_pattern() {
        // Ignore rules cover every entry kind: an excluded symlink stays out
        // of the manifest, an excluded directory with nothing re-included
        // below it gets no entry, and the gitignore walk keeps skipping
        // symlinks entirely.
        let dir = tempfile::tempdir().expect("tempdir");
        std::fs::create_dir(dir.path().join("logs")).unwrap();
        std::fs::write(dir.path().join("logs/app.log"), b"log").unwrap();
        std::fs::write(dir.path().join("app.py"), b"app").unwrap();
        std::os::unix::fs::symlink("app.py", dir.path().join("drop.link")).unwrap();
        let walked = walk_dir_files(
            "contextDir",
            dir.path(),
            &docker_context_rules(b"logs\ndrop.link\n!nothing\n", &[]),
            MAX_DOCKERFILE_CONTEXT_FILES,
        )
        .expect("walk");
        let files: Vec<_> = walked
            .files
            .iter()
            .map(|f| f.relative_path.clone())
            .collect();
        assert_eq!(files, ["app.py"]);
        assert!(walked.dirs.is_empty(), "{:?}", walked.dirs);
        assert!(walked.symlinks.is_empty(), "{:?}", walked.symlinks);

        let walked = walk_dir_files(
            "addLocalDir",
            dir.path(),
            &DirWalkRules::Gitignore {
                ignore: Vec::new(),
                ignore_file: None,
            },
            MAX_LOCAL_DIR_FILES,
        )
        .expect("walk");
        assert!(walked.dirs.is_empty(), "{:?}", walked.dirs);
        assert!(walked.symlinks.is_empty(), "{:?}", walked.symlinks);
    }

    #[test]
    fn dockerfile_context_walk_skips_sockets_and_rejects_pipes() {
        let dir = tempfile::tempdir().expect("tempdir");
        std::fs::write(dir.path().join("app.py"), b"app").unwrap();
        let _listener = std::os::unix::net::UnixListener::bind(dir.path().join("live.sock"))
            .expect("bind test socket");
        // A socket stays silently out of the manifest, like docker's own
        // context handling.
        let walked = walk_dir_files(
            "contextDir",
            dir.path(),
            &docker_context_rules(b"", &[]),
            MAX_DOCKERFILE_CONTEXT_FILES,
        )
        .expect("a socket must not fail the walk");
        let files: Vec<_> = walked
            .files
            .iter()
            .map(|f| f.relative_path.clone())
            .collect();
        assert_eq!(files, ["app.py"]);

        // A named pipe fails the walk: it has no hashable content, and
        // skipping it would silently diverge from a local docker build.
        let status = std::process::Command::new("mkfifo")
            .arg(dir.path().join("events.fifo"))
            .status()
            .expect("run mkfifo");
        assert!(status.success());
        let err = walk_dir_files(
            "contextDir",
            dir.path(),
            &docker_context_rules(b"", &[]),
            MAX_DOCKERFILE_CONTEXT_FILES,
        )
        .unwrap_err()
        .to_string();
        assert!(err.contains("events.fifo"), "{err}");
        assert!(err.contains("named pipe or device node"), "{err}");

        // An ignored pipe never ships, so it does not fail the walk either.
        walk_dir_files(
            "contextDir",
            dir.path(),
            &docker_context_rules(b"events.fifo\n", &[]),
            MAX_DOCKERFILE_CONTEXT_FILES,
        )
        .expect("an ignored pipe must not fail the walk");

        // The addLocalDir walk keeps its shipped files-only behavior.
        let walked = walk_dir_files(
            "addLocalDir",
            dir.path(),
            &DirWalkRules::Gitignore {
                ignore: Vec::new(),
                ignore_file: None,
            },
            MAX_LOCAL_DIR_FILES,
        )
        .expect("addLocalDir silently skips special files");
        let files: Vec<_> = walked
            .files
            .iter()
            .map(|f| f.relative_path.clone())
            .collect();
        assert_eq!(files, ["app.py"]);
    }

    #[test]
    fn dockerfile_context_walk_rejects_setuid_setgid_sticky_bits() {
        use std::os::unix::fs::PermissionsExt;
        let dir = tempfile::tempdir().expect("tempdir");
        std::fs::write(dir.path().join("tool"), b"#!/bin/sh\n").unwrap();
        std::fs::set_permissions(
            dir.path().join("tool"),
            std::fs::Permissions::from_mode(0o4755),
        )
        .unwrap();
        let err = walk_dir_files(
            "contextDir",
            dir.path(),
            &docker_context_rules(b"", &[]),
            MAX_DOCKERFILE_CONTEXT_FILES,
        )
        .unwrap_err()
        .to_string();
        assert!(err.contains("tool"), "{err}");
        assert!(err.contains("setuid, setgid, or sticky"), "{err}");

        // The addLocalDir walk keeps its shipped behavior: the manifest
        // records only the lower bits, silently.
        let walked = walk_dir_files(
            "addLocalDir",
            dir.path(),
            &DirWalkRules::Gitignore {
                ignore: Vec::new(),
                ignore_file: None,
            },
            MAX_LOCAL_DIR_FILES,
        )
        .expect("addLocalDir strips special mode bits silently");
        assert_eq!(walked.files[0].mode, 0o755);

        // A setgid directory is refused the same way.
        std::fs::set_permissions(
            dir.path().join("tool"),
            std::fs::Permissions::from_mode(0o755),
        )
        .unwrap();
        std::fs::create_dir(dir.path().join("shared")).unwrap();
        std::fs::set_permissions(
            dir.path().join("shared"),
            std::fs::Permissions::from_mode(0o2775),
        )
        .unwrap();
        let err = walk_dir_files(
            "contextDir",
            dir.path(),
            &docker_context_rules(b"", &[]),
            MAX_DOCKERFILE_CONTEXT_FILES,
        )
        .unwrap_err()
        .to_string();
        assert!(err.contains("shared"), "{err}");
        assert!(err.contains("setuid, setgid, or sticky"), "{err}");
    }

    #[test]
    fn dockerfile_context_walk_rejects_mode_000() {
        use std::os::unix::fs::PermissionsExt;
        let dir = tempfile::tempdir().expect("tempdir");
        std::fs::write(dir.path().join("locked.bin"), b"x").unwrap();
        std::fs::set_permissions(
            dir.path().join("locked.bin"),
            std::fs::Permissions::from_mode(0o000),
        )
        .unwrap();
        let err = walk_dir_files(
            "contextDir",
            dir.path(),
            &docker_context_rules(b"", &[]),
            MAX_DOCKERFILE_CONTEXT_FILES,
        )
        .unwrap_err()
        .to_string();
        assert!(err.contains("locked.bin"), "{err}");
        assert!(err.contains("mode 000"), "{err}");

        // A directory is refused the same way, before the walk descends
        // into it: listing a 000 directory needs privilege, so the mode
        // check has to come first for the error to name the real problem.
        std::fs::set_permissions(
            dir.path().join("locked.bin"),
            std::fs::Permissions::from_mode(0o644),
        )
        .unwrap();
        std::fs::create_dir(dir.path().join("vault")).unwrap();
        std::fs::set_permissions(
            dir.path().join("vault"),
            std::fs::Permissions::from_mode(0o000),
        )
        .unwrap();
        let err = walk_dir_files(
            "contextDir",
            dir.path(),
            &docker_context_rules(b"", &[]),
            MAX_DOCKERFILE_CONTEXT_FILES,
        )
        .unwrap_err()
        .to_string();
        assert!(err.contains("vault"), "{err}");
        assert!(err.contains("mode 000"), "{err}");
        // Restore traversal so tempdir cleanup can remove the tree.
        std::fs::set_permissions(
            dir.path().join("vault"),
            std::fs::Permissions::from_mode(0o755),
        )
        .unwrap();
    }

    #[test]
    fn extended_dockerignore_appends_patterns_as_lines() {
        let patterns = vec!["!keep.log".to_string(), "extra/".to_string()];
        assert_eq!(
            extended_dockerignore(b"*.log\n", &patterns),
            b"*.log\n!keep.log\nextra/\n"
        );
        // A file missing its trailing newline still gets each pattern on its
        // own line.
        assert_eq!(
            extended_dockerignore(b"*.log", &patterns),
            b"*.log\n!keep.log\nextra/\n"
        );
        assert_eq!(
            extended_dockerignore(b"", &patterns),
            b"!keep.log\nextra/\n"
        );
        assert_eq!(extended_dockerignore(b"*.log\n", &[]), b"*.log\n");
    }
}