alien-core 3.3.25

//! Sandbox resource for running untrusted code in an isolated environment.
//!
//! The declaration provisions a durable parent, and the application creates and destroys
//! individual sandboxes through its binding at runtime.
//!
//! The capability set differs per platform and is published rather than assumed. Calling an
//! unsupported capability is a typed error naming both the platform and the capability, so a
//! portable application can branch on `SandboxCapabilities` before it calls.

use crate::error::{ErrorData, Result};
use crate::resource::{ResourceDefinition, ResourceOutputsDefinition, ResourceRef, ResourceType};
use crate::resources::ToolchainConfig;
use crate::Platform;
use alien_error::AlienError;
use bon::Builder;
use serde::{Deserialize, Serialize};
use std::any::Any;
use std::fmt::Debug;

/// Specifies where the sandbox's root filesystem comes from.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(rename_all = "camelCase", tag = "type")]
pub enum SandboxCode {
    /// A prebuilt container image used as the sandbox root filesystem.
    #[serde(rename_all = "camelCase")]
    Image {
        /// Image reference (e.g. `ubuntu:24.04`, `ghcr.io/myorg/sandbox:latest`).
        ///
        /// Two backends narrow it in opposite directions: AWS wants an `s3://` bundle, Azure a
        /// bare catalog name such as `ubuntu`. Each refuses the other's shape while planning.
        image: String,
    },
    /// A Dockerfile `alien build` builds into the sandbox's base image.
    ///
    /// AWS only, and docker only: the base image is a root filesystem, not a binary laid on one.
    /// `alien release` pushes it and the bundle layers the sandbox agent on afterwards.
    #[serde(rename_all = "camelCase")]
    Source {
        /// The source directory to build from
        src: String,
        /// Toolchain configuration with type-safe options
        toolchain: ToolchainConfig,
    },
}

/// Hard ceilings enforced on a sandbox.
///
/// These are limits, not scheduling requests. Untrusted code does not respect a hint, so every
/// field is enforced by the platform and a platform that cannot enforce one is rejected at plan
/// time rather than silently ignoring it.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct SandboxLimits {
    /// CPU ceiling in cores or millicores (e.g. `"1"`, `"500m"`)
    pub cpu: String,
    /// Memory ceiling (e.g. `"2Gi"`, `"512Mi"`)
    pub memory: String,
    /// Disk ceiling (e.g. `"20Gi"`)
    pub disk: String,
    /// Maximum number of processes, which bounds fork bombs.
    ///
    /// Optional because only a container runtime has the primitive: Kubernetes sets a pid ceiling
    /// per node, not per pod, and neither AWS MicroVMs nor Azure sandboxes expose one. Declaring
    /// it on a platform that cannot apply it is refused at plan time.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub max_processes: Option<u32>,
}

/// One of the five sizes a Lambda MicroVM can be built at.
///
/// AWS has no ceiling knob: `minimumMemoryInMiB` sets a *baseline* and a running MicroVM bursts
/// vertically to four times it with no way to opt out. A declared ceiling is therefore honoured by
/// picking the tier whose **peak** stays inside it, not the tier whose baseline matches it.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct MicrovmTier {
    /// What `minimumMemoryInMiB` is set to.
    pub baseline_memory_mib: i64,
    /// The most memory the MicroVM can reach, in MiB.
    pub peak_memory_mib: i64,
    /// The most vCPU the MicroVM can reach.
    pub peak_vcpu: u32,
    /// The most disk the MicroVM can use, in MiB.
    pub max_disk_mib: i64,
}

/// The published sizes, smallest first. Baseline memory to vCPU is 2 GB per vCPU, peak is four
/// times baseline, and disk is fixed per tier rather than independently selectable.
/// Longest life AWS will run a MicroVM for, from `RunMicrovm`'s `maximumDurationInSeconds`.
const AWS_MAX_LIFETIME_SECONDS: u32 = 28_800;

/// Azure's sandbox sizing rule, quoted from the data plane's own refusal of an oversized request:
/// *CPU must be n×250m for n=1..64 (0.25–16 cores); Memory ≤ cores × 2Gi; Disk ≤ cores × 20Gi*.
const AZURE_CPU_STEP_MILLICORES: i64 = 250;
const AZURE_MAX_CPU_MILLICORES: i64 = 16_000;
const AZURE_MEMORY_MIB_PER_CORE: i64 = 2 * 1024;
const AZURE_DISK_MIB_PER_CORE: i64 = 20 * 1024;

const MICROVM_TIERS: &[MicrovmTier] = &[
    MicrovmTier {
        baseline_memory_mib: 512,
        peak_memory_mib: 2048,
        peak_vcpu: 1,
        max_disk_mib: 8192,
    },
    MicrovmTier {
        baseline_memory_mib: 1024,
        peak_memory_mib: 4096,
        peak_vcpu: 2,
        max_disk_mib: 8192,
    },
    MicrovmTier {
        baseline_memory_mib: 2048,
        peak_memory_mib: 8192,
        peak_vcpu: 4,
        max_disk_mib: 8192,
    },
    MicrovmTier {
        baseline_memory_mib: 4096,
        peak_memory_mib: 16384,
        peak_vcpu: 8,
        max_disk_mib: 16384,
    },
    MicrovmTier {
        baseline_memory_mib: 8192,
        peak_memory_mib: 32768,
        peak_vcpu: 16,
        max_disk_mib: 32768,
    },
];

/// Outbound network policy for a sandbox.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(rename_all = "camelCase", tag = "mode")]
pub enum SandboxEgress {
    /// No outbound network access.
    ///
    /// Routed traffic only. Link-local is not outbound and no backend's egress control reaches
    /// it, so this is not a boundary against instance metadata.
    ///
    /// Nor, on AWS, against DNS: while the connector VPC has DNS support on, a session resolves
    /// names through that VPC's resolver, which no security group filters, so a query name can
    /// carry data out.
    Deny,
    /// Unrestricted outbound access to the public internet, and none to private ranges or the
    /// deployment's own network.
    ///
    /// Link-local carries the same exception as `Deny`. AWS and Kubernetes deliver both halves.
    /// Azure and GCP deliver the first only: one matches host patterns and the other is a single
    /// switch, so neither can name an address range to exclude.
    Allow,
    /// Outbound access only to the listed hostnames.
    ///
    /// Azure alone expresses it: its egress proxy matches on host pattern. The others filter by
    /// CIDR or carry a single switch, and both would approximate the list rather than keep it.
    #[serde(rename_all = "camelCase")]
    AllowDomains {
        /// Hostnames the sandbox may reach
        domains: Vec<String>,
    },
}

impl SandboxEgress {
    /// The single outbound switch for a backend that has no host matcher, or `None` for a mode a
    /// boolean cannot carry.
    ///
    /// `AllowDomains` needs a host list, so it maps to nothing and each caller refuses it in its
    /// own error naming the sandbox. One source for what a mode means, so a template and a sandbox
    /// cannot disagree on it.
    pub fn internet_access_switch(&self) -> Option<bool> {
        match self {
            SandboxEgress::Allow => Some(true),
            SandboxEgress::Deny => Some(false),
            SandboxEgress::AllowDomains { .. } => None,
        }
    }
}

/// How long a sandbox may live and when it is paused.
///
/// Declaration-time ceilings, not per-request values: every sandbox created through this
/// declaration's binding is held to them, whatever a caller asks for at runtime.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct SandboxLifecyclePolicy {
    /// Wall-clock ceiling on a single sandbox, after which the platform terminates it.
    ///
    /// Optional because not every backend has the primitive: Kubernetes has
    /// `activeDeadlineSeconds` and AWS `maximumDurationInSeconds`, while neither Azure nor Local
    /// expose one, so declaring a ceiling there is refused at plan time rather than accepted and
    /// never applied. AWS caps it at 8 hours.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub max_lifetime_seconds: Option<u32>,
    /// Idle period after which the sandbox is paused, where the platform supports it
    #[serde(skip_serializing_if = "Option::is_none")]
    pub idle_pause_seconds: Option<u32>,
}

/// What a platform's sandbox backend can actually do.
///
/// Published so portable code can branch before calling rather than discovering a gap through
/// an error. Every field here corresponds to a capability that at least one platform lacks;
/// create, exec and terminate are the guaranteed floor and are therefore not listed.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct SandboxCapabilities {
    /// Files can be moved in and out of a sandbox
    pub files: bool,
    /// A later call can reach a sandbox created by an earlier one
    pub reconnect: bool,
    /// A command can be started, polled and cancelled across separate calls, so it outlives the
    /// one that started it. False where nothing inside the sandbox owns the process in between.
    pub jobs: bool,
    /// An authenticated, port-scoped capability to reach a service inside the sandbox
    pub preview: bool,
    /// Sandbox state can be paused and resumed
    pub pause_resume: bool,
    /// A sandbox's full state can be captured and used to create another
    pub snapshot: bool,
    /// Egress can be restricted to a hostname allowlist
    pub domain_egress_rules: bool,
    /// Whether a declared `deny` is actually enforced, rather than accepted and dropped.
    /// It covers routed traffic; the `deny` mode says where DNS still resolves.
    pub egress_deny: bool,
    /// The platform enforces the declared cpu, memory and disk ceilings
    pub enforced_limits: bool,
    /// The platform can cap how many processes a sandbox runs
    pub process_limit: bool,
    /// The platform terminates a sandbox at a declared wall-clock deadline
    pub sandbox_lifetime: bool,
    /// A command runs in its own PID namespace and cannot see or signal the agent's processes.
    ///
    /// Only where an agent runs as root. Creating the namespace needs `CAP_SYS_ADMIN`, and the
    /// Kubernetes sandbox pod drops every capability — which is also what denies `ptrace` by
    /// construction, so granting it there would remove a lock to add one.
    pub supervisor_pid_namespace: bool,
    /// The process supervising a command is a different identity from the command.
    ///
    /// False where a command runs as the agent's own user: it can then read the supervisor's
    /// environment and signal it. Separate from `supervisorPidNamespace`, which is about
    /// visibility rather than identity — a backend can have one without the other.
    pub supervisor_isolation: bool,
}

impl SandboxCapabilities {
    /// Returns what the given platform's sandbox backend supports.
    ///
    /// Errors for platforms with no sandbox backend, rather than returning an all-false set —
    /// "every capability is missing" and "this platform has no sandboxes" are different
    /// conditions and an application should not have to tell them apart by inspection.
    pub fn for_platform(platform: Platform) -> Result<Self> {
        match platform {
            Platform::Aws => Ok(Self {
                files: true,
                reconnect: true,
                jobs: true,
                preview: true,
                pause_resume: true,
                snapshot: false,
                domain_egress_rules: false,
                // Routed egress only: while the connector VPC has DNS support on, a session
                // resolves names through its resolver, which no security group filters.
                egress_deny: true,
                enforced_limits: true,
                // Nothing in the API bounds process count.
                process_limit: false,
                // `maximumDurationInSeconds` on `RunMicrovm`, which Lambda enforces by
                // terminating the MicroVM. Capped at 8 hours by the service.
                sandbox_lifetime: true,
                // Measured, not assumed: the agent inside a Lambda MicroVM runs as uid 0 with
                // `CapEff: 00000000a80425fb`, the standard container default set, which excludes
                // `CAP_SYS_ADMIN`. It can drop privilege (`CAP_SETUID`/`CAP_SETGID` are held) and
                // it cannot create a namespace. No backend offers this today.
                supervisor_pid_namespace: false,
                // The agent runs as uid 0 and `setuid`s the command to uid 60000, so the command
                // runs under a different identity than the process supervising it.
                supervisor_isolation: true,
            }),
            Platform::Azure => Ok(Self::azure()),
            Platform::Gcp => Ok(Self::gcp_agent_platform()),
            // Preview needs a gateway that validates a sandbox-and-port capability, which this
            // backend has none of.
            Platform::Kubernetes => Ok(Self {
                files: true,
                reconnect: true,
                jobs: true,
                preview: false,
                pause_resume: false,
                snapshot: false,
                domain_egress_rules: false,
                egress_deny: true,
                enforced_limits: true,
                // A pid ceiling is a kubelet setting per node, not a pod field.
                process_limit: false,
                // `activeDeadlineSeconds` on the pod, which the kubelet enforces.
                sandbox_lifetime: true,
                // The pod drops every capability, including the `CAP_SYS_ADMIN` the agent would
                // need to unshare. That is also what denies `ptrace`, so this stays false rather
                // than the pod being weakened to make it true.
                supervisor_pid_namespace: false,
                // The pod pins one uid (`run_as_user: 65534` on both pod and container) with
                // `capabilities.drop: [ALL]` and `allow_privilege_escalation: false`, so no
                // process can setuid to split the command off from a supervisor. No uid split is
                // possible, so none exists.
                supervisor_isolation: false,
            }),
            Platform::Local => Ok(Self {
                files: true,
                reconnect: true,
                // Nothing runs inside the sandbox: the manager drives Docker from outside it.
                jobs: false,
                preview: true,
                pause_resume: false,
                snapshot: false,
                domain_egress_rules: false,
                egress_deny: true,
                enforced_limits: true,
                // Docker's `--pids-limit`.
                process_limit: true,
                sandbox_lifetime: false,
                // Local has no in-sandbox agent: the manager drives Docker from outside, so
                // there is no supervisor inside the sandbox to isolate from.
                supervisor_pid_namespace: false,
                // The supervisor is the manager on the host, outside the container entirely, and
                // `docker exec` runs the command as the workload uid — a different identity by
                // construction.
                supervisor_isolation: true,
            }),
            Platform::Machines | Platform::Test => {
                Err(AlienError::new(ErrorData::SandboxPlatformUnsupported {
                    platform: platform.to_string(),
                }))
            }
        }
    }

    /// What the Azure sandbox backend supports; the body of the `Platform::Azure` arm.
    pub fn azure() -> Self {
        Self {
            files: true,
            reconnect: true,
            // No Alien process runs inside the sandbox to own a command between two calls.
            jobs: false,
            // A sandbox port carries a URL and an auth config, and the auth config offers two
            // things: anonymous, or Entra ID with an allowlist of human email addresses.
            // Neither is a credential scoped to a port for a fixed time, which is what a
            // preview capability is. Returning the anonymous URL would publish the port.
            preview: false,
            pause_resume: true,
            // False for a client reason, not a cloud one: this client has no snapshot call, and
            // `CreateSandboxRequest` has no field to consume the id it would return. Also
            // unclaimed: Microsoft does not garbage-collect snapshots, so an id is a bill that grows.
            snapshot: false,
            domain_egress_rules: true,
            egress_deny: true,
            // Enforced inside the sandbox, not at create: an over-allocation raises `MemoryError`
            // while the sandbox keeps running. `azure_sandbox_limits` checks the continuous
            // sizing rule at plan time instead of matching a tier.
            enforced_limits: true,
            process_limit: false,
            // Auto-suspend and auto-delete exist; a wall-clock ceiling does not. Accepting
            // `maxLifetimeSeconds` here would be the silent no-op the capability set exists
            // to prevent, so this is a decision rather than a gap.
            sandbox_lifetime: false,
            // No Alien process inside an Azure sandbox, so there is no supervisor to isolate.
            supervisor_pid_namespace: false,
            // No Alien process runs the command at all — the platform's own data plane does,
            // so there is no separate supervisor identity to speak of.
            supervisor_isolation: false,
        }
    }

    /// What the GCP Agent Platform sandbox backend supports; the body of the `Platform::Gcp` arm.
    pub fn gcp_agent_platform() -> Self {
        Self {
            // Agent file operations move over the sandbox envelope.
            files: true,
            // Reaching a sandbox across processes is safe because `generation` is derived from the
            // container boot id read through the agent's health op, so a caller detects a container
            // replaced under a stable sandbox name rather than reconnecting to a blank one.
            reconnect: true,
            jobs: true,
            // No method mints a port-scoped ingress capability; the only ingress is `:execute`.
            preview: false,
            // `:pause` and `:resume` preserve the running container.
            pause_resume: true,
            // The create path never sends `sandbox_environment_snapshot`, so no sandbox state is
            // reachable through the trait; declared false until the client carries it.
            snapshot: false,
            // Egress is shaped by VPC and DNS peering, which is not a hostname allowlist.
            domain_egress_rules: false,
            // A declared `deny` blocks both routed egress and DNS.
            egress_deny: true,
            // The declared ceilings are enforced, but by terminating the sandbox on breach rather
            // than by refusing the allocation — a caller reading `true` should expect the sandbox
            // to die, not a clean error at the point of the request.
            enforced_limits: true,
            // No ceiling on process count is observed.
            process_limit: false,
            // `ttl` maps to a sandbox `expireTime` the platform terminates at.
            sandbox_lifetime: true,
            // No PID-namespace isolation between the command and anything supervising it.
            supervisor_pid_namespace: false,
            // No separate supervisor identity: the command is not run under a different identity
            // than the process supervising it.
            supervisor_isolation: false,
        }
    }

    /// Returns a typed error if the named capability is absent on this platform.
    pub fn require(&self, capability: SandboxCapability, platform: Platform) -> Result<()> {
        let available = match capability {
            SandboxCapability::Files => self.files,
            SandboxCapability::Reconnect => self.reconnect,
            SandboxCapability::Jobs => self.jobs,
            SandboxCapability::Preview => self.preview,
            SandboxCapability::PauseResume => self.pause_resume,
            SandboxCapability::Snapshot => self.snapshot,
            SandboxCapability::DomainEgressRules => self.domain_egress_rules,
            SandboxCapability::EgressDeny => self.egress_deny,
            SandboxCapability::EnforcedLimits => self.enforced_limits,
            SandboxCapability::ProcessLimit => self.process_limit,
            SandboxCapability::SandboxLifetime => self.sandbox_lifetime,
            SandboxCapability::SupervisorPidNamespace => self.supervisor_pid_namespace,
            SandboxCapability::SupervisorIsolation => self.supervisor_isolation,
        };

        if available {
            return Ok(());
        }

        Err(AlienError::new(ErrorData::SandboxCapabilityUnsupported {
            capability: capability.as_str().to_string(),
            platform: platform.to_string(),
        }))
    }
}

/// Names a single sandbox capability, so an unsupported call can report which one it needed.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(rename_all = "camelCase")]
pub enum SandboxCapability {
    /// Moving files in and out of a sandbox
    Files,
    /// Reaching a sandbox created by an earlier call
    Reconnect,
    /// Starting, polling and cancelling a command across separate calls
    Jobs,
    /// An authenticated, port-scoped ingress capability
    Preview,
    /// Pausing and resuming sandbox state
    PauseResume,
    /// Capturing full sandbox state
    Snapshot,
    /// Restricting egress to a hostname allowlist
    DomainEgressRules,
    /// Refusing outbound access when a sandbox declares none
    EgressDeny,
    /// Platform-enforced resource ceilings
    EnforcedLimits,
    /// A ceiling on the number of processes a sandbox may run
    ProcessLimit,
    /// A wall-clock ceiling on a sandbox, applied by the platform rather than by a caller
    SandboxLifetime,
    /// A command runs in its own PID namespace, isolated from the agent supervising it
    SupervisorPidNamespace,
    /// A command runs under a different identity than the process supervising it
    SupervisorIsolation,
}

impl SandboxCapability {
    /// Returns the stable identifier used in errors and capability queries.
    pub fn as_str(&self) -> &'static str {
        match self {
            Self::Files => "files",
            Self::Reconnect => "reconnect",
            Self::Jobs => "jobs",
            Self::Preview => "preview",
            Self::PauseResume => "pauseResume",
            Self::Snapshot => "snapshot",
            Self::DomainEgressRules => "domainEgressRules",
            Self::EgressDeny => "egressDeny",
            Self::EnforcedLimits => "enforcedLimits",
            Self::ProcessLimit => "processLimit",
            Self::SandboxLifetime => "sandboxLifetime",
            Self::SupervisorPidNamespace => "supervisorPidNamespace",
            Self::SupervisorIsolation => "supervisorIsolation",
        }
    }
}

/// An isolated environment for running untrusted code, created at runtime.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Builder)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
#[builder(start_fn = new)]
pub struct Sandbox {
    /// Identifier for the sandbox. Must contain only alphanumeric characters, hyphens, and
    /// underscores ([A-Za-z0-9-_]). Maximum 64 characters.
    #[builder(start_fn)]
    pub id: String,
    /// Where the sandbox's root filesystem comes from
    pub code: SandboxCode,
    /// Private ECR base image an AWS build pulls; `code.image` names only the S3 bundle, so this is
    /// what the cross-account grant opens. Live only: the grant needs the customer account,
    /// which registration reports. Absent means the base image is pulled anonymously.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub private_base_image: Option<String>,
    /// Enforced resource ceilings.
    ///
    /// Optional because not every platform can enforce them, and a declaration that names none
    /// takes the platform's own defaults. Naming them on a platform that cannot enforce them is
    /// rejected at plan time rather than silently ignored.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub limits: Option<SandboxLimits>,
    /// Outbound network policy
    pub egress: SandboxEgress,
    /// Sandbox lifetime ceiling and idle behaviour
    pub lifecycle: SandboxLifecyclePolicy,
    /// Ports eligible for a preview capability. An application reaches its sandbox through the
    /// provider, so it cannot widen its own ingress at runtime; a holder of a remote binding's
    /// credentials is bounded by no port condition, which is why a remote sandbox declares none.
    #[builder(default)]
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub preview_ports: Vec<u16>,
}

/// Whether the artifact being rendered restricts which network modes it accepts.
///
/// Cloud setup needs explicit subnets for restricted sandbox connectors. Kubernetes targets do
/// not emit these cloud backends.
pub fn restricts_network_mode(stack: &crate::Stack, targets_kubernetes: bool) -> bool {
    !targets_kubernetes && stack_needs_named_subnets_at_setup(stack)
}

/// Whether any setup-owned resource forces setup to name subnets.
///
/// Restricted sandbox connectors require subnet IDs. Callers rendering an artifact want
/// [`restricts_network_mode`] instead:
/// this one answers for the declaration, which on a Kubernetes target is not what gets emitted.
pub fn stack_needs_named_subnets_at_setup(stack: &crate::Stack) -> bool {
    stack.resources().any(|(_resource_id, resource)| {
        resource
            .config
            .downcast_ref::<Sandbox>()
            .is_some_and(|sandbox| !matches!(sandbox.egress, SandboxEgress::Allow))
    })
}

impl Sandbox {
    /// The resource type identifier for Sandbox
    pub const RESOURCE_TYPE: ResourceType = ResourceType::from_static("sandbox");

    /// Returns the sandbox's unique identifier.
    pub fn id(&self) -> &str {
        &self.id
    }

    /// The declared ceilings, or the defaults a platform applies when none were named.
    ///
    /// Backends want a concrete set: a sandbox with no declared ceilings still runs inside
    /// whatever the platform gives it, and a backend that had to branch on `None` would end up
    /// inventing its own default anyway.
    pub fn resolved_limits(&self) -> SandboxLimits {
        self.limits.clone().unwrap_or_else(default_limits)
    }

    /// Validates the declaration against what the target platform can enforce.
    ///
    /// Runs at plan time so an unenforceable limit or an unsupported egress mode fails before
    /// anything is provisioned, rather than at the first exec.
    pub fn validate_for_platform(&self, platform: Platform) -> Result<()> {
        let capabilities = SandboxCapabilities::for_platform(platform)?;

        // `alien build` builds an AWS sandbox's base image, so source is a declaration there and
        // the emitters refuse it only if it reaches them unbuilt. Everywhere else the image is
        // pulled as declared, and an empty image string would schedule a pod that can never run.
        if matches!(&self.code, SandboxCode::Source { .. }) && platform != Platform::Aws {
            return Err(AlienError::new(ErrorData::SandboxLimitInvalid {
                resource_id: self.id.clone(),
                field: "code".to_string(),
                value: "source".to_string(),
                reason: format!(
                    "no sandbox backend builds an image from source on {platform}; give \
                     code.image a prebuilt reference"
                ),
            }));
        }

        // Elsewhere `code.image` is pulled directly, so a second reference would be a grant
        // nothing reads.
        if self.private_base_image.is_some() && platform != Platform::Aws {
            return Err(AlienError::new(ErrorData::SandboxCapabilityUnsupported {
                capability: "privateBaseImage".to_string(),
                platform: platform.to_string(),
            }));
        }

        // Read before the limits, because the image is declared whether or not any are.
        if platform == Platform::Azure {
            self.azure_catalog_image()?;
        }

        let Some(limits) = self.limits.as_ref() else {
            // Nothing declared, so nothing to enforce and nothing to reject.
            return self.validate_capabilities(&capabilities, platform);
        };

        validate_quantity(&self.id, "cpu", &limits.cpu)?;
        validate_quantity(&self.id, "memory", &limits.memory)?;
        validate_quantity(&self.id, "disk", &limits.disk)?;

        if let Some(max_processes) = limits.max_processes {
            if max_processes == 0 {
                return Err(AlienError::new(ErrorData::SandboxLimitInvalid {
                    resource_id: self.id.clone(),
                    field: "maxProcesses".to_string(),
                    value: "0".to_string(),
                    reason: "a sandbox that may run no processes cannot run code".to_string(),
                }));
            }
            capabilities.require(SandboxCapability::ProcessLimit, platform)?;
        }

        // Declaring limits a platform ignores is worse than not declaring them: the stack reads
        // as bounded while the sandbox is not.
        capabilities.require(SandboxCapability::EnforcedLimits, platform)?;

        if platform == Platform::Azure {
            self.azure_sandbox_limits()?;
        }

        if platform == Platform::Aws {
            // Refused here rather than at emit so a customer sees it while planning, and so both
            // package formats inherit the same answer.
            self.microvm_tier()?;

            // The ceiling is Lambda's, and it rejects the run rather than clamping — so a value
            // outside it would pass planning, render into the package, and fail at the first
            // sandbox. Kubernetes takes the same field with no such bound, which is why this
            // sits under the AWS gate rather than on the type.
            if let Some(seconds) = self.lifecycle.max_lifetime_seconds {
                if !(1..=AWS_MAX_LIFETIME_SECONDS).contains(&seconds) {
                    return Err(AlienError::new(ErrorData::SandboxLimitInvalid {
                        resource_id: self.id.clone(),
                        field: "maxLifetimeSeconds".to_string(),
                        value: seconds.to_string(),
                        reason: format!(
                            "AWS runs a MicroVM for between 1 and \
                             {AWS_MAX_LIFETIME_SECONDS} seconds"
                        ),
                    }));
                }
            }
        }

        self.validate_capabilities(&capabilities, platform)
    }

    /// The catalog disk image Azure creates a sandbox from.
    ///
    /// Azure names a public catalog entry rather than pulling a reference, so a registry path,
    /// tag or digest has nowhere to go. An allowlist, because the answer to "what else could be
    /// in there" is a name the data plane rejects at the first sandbox, long after the apply.
    pub fn azure_catalog_image(&self) -> Result<&str> {
        let refused = |value: &str, reason: &str| {
            AlienError::new(ErrorData::SandboxLimitInvalid {
                resource_id: self.id.clone(),
                field: "code.image".to_string(),
                value: value.to_string(),
                reason: reason.to_string(),
            })
        };

        let SandboxCode::Image { image } = &self.code else {
            return Err(AlienError::new(ErrorData::SandboxLimitInvalid {
                resource_id: self.id.clone(),
                field: "code".to_string(),
                value: "source".to_string(),
                reason: "no sandbox backend builds an image from source yet".to_string(),
            }));
        };

        let image = image.trim();
        if image.is_empty() {
            return Err(refused(image, "a sandbox has to name an image"));
        }
        if !image
            .chars()
            .all(|c| c.is_ascii_alphanumeric() || matches!(c, '.' | '_' | '-'))
        {
            return Err(refused(
                image,
                "Azure creates a sandbox from a public catalog disk image, so code.image must be \
                 a bare catalog name such as 'ubuntu'",
            ));
        }
        Ok(image)
    }

    /// Checks the declared ceilings against Azure's sizing rule (the `AZURE_*` constants above).
    /// Refused at plan time, like [`Self::microvm_tier`], so a bad value is a declaration to fix
    /// rather than a runtime fault at create.
    pub fn azure_sandbox_limits(&self) -> Result<()> {
        let Some(limits) = self.limits.as_ref() else {
            // Nothing declared means the binding substitutes Alien's own default sizing, which is
            // inside the rule — asserted where those constants live, since this cannot see them.
            return Ok(());
        };

        let refused = |field: &str, value: &str, reason: &str| {
            AlienError::new(ErrorData::SandboxLimitInvalid {
                resource_id: self.id.clone(),
                field: field.to_string(),
                value: value.to_string(),
                reason: reason.to_string(),
            })
        };

        let cpu_millicores = millicores(&limits.cpu)
            .ok_or_else(|| refused("cpu", &limits.cpu, "expected cores or millicores"))?;

        // The multiple is checked, not just the range: `333m` sits inside 0.25–16 cores and is
        // still refused on the wire, so a bounds-only check would pass a declaration that fails
        // at create.
        if cpu_millicores % AZURE_CPU_STEP_MILLICORES != 0
            || !(AZURE_CPU_STEP_MILLICORES..=AZURE_MAX_CPU_MILLICORES).contains(&cpu_millicores)
        {
            return Err(refused(
                "cpu",
                &limits.cpu,
                "Azure allocates cpu in steps of 250m from 250m to 16000m",
            ));
        }

        // Both ceilings are derived from the cpu, so they cannot be checked before it is known.
        let memory_ceiling_mib = cpu_millicores * AZURE_MEMORY_MIB_PER_CORE / 1000;
        let disk_ceiling_mib = cpu_millicores * AZURE_DISK_MIB_PER_CORE / 1000;

        let memory_mib = quantity_mib(&limits.memory)
            .ok_or_else(|| refused("memory", &limits.memory, "Azure sizes memory in whole MiB"))?;
        if memory_mib > memory_ceiling_mib {
            return Err(refused(
                "memory",
                &limits.memory,
                &format!(
                    "Azure allows at most 2Gi of memory per core, or {memory_ceiling_mib}Mi \
                          at the declared cpu"
                ),
            ));
        }

        let disk_mib = quantity_mib(&limits.disk)
            .ok_or_else(|| refused("disk", &limits.disk, "Azure sizes disk in whole MiB"))?;
        if disk_mib > disk_ceiling_mib {
            return Err(refused(
                "disk",
                &limits.disk,
                &format!(
                    "Azure allows at most 20Gi of disk per core, or {disk_ceiling_mib}Mi at \
                          the declared cpu"
                ),
            ));
        }

        Ok(())
    }

    /// The MicroVM size that keeps every declared ceiling, or why none does.
    ///
    /// AWS sizes are discrete and a running MicroVM bursts to four times its baseline, so the
    /// only tier that honours a ceiling is one whose peak fits inside it. A declaration no tier
    /// satisfies is refused: shipping the nearest size would give the customer a sandbox that
    /// exceeds the bound they wrote down.
    pub fn microvm_tier(&self) -> Result<MicrovmTier> {
        let Some(limits) = self.limits.as_ref() else {
            // Nothing declared: AWS's own default baseline, which is also `default_limits`.
            return Ok(MICROVM_TIERS[2]);
        };

        let memory_mib = quantity_mib(&limits.memory).ok_or_else(|| {
            AlienError::new(ErrorData::SandboxLimitInvalid {
                resource_id: self.id.clone(),
                field: "memory".to_string(),
                value: limits.memory.clone(),
                reason: "AWS sizes a MicroVM in whole MiB".to_string(),
            })
        })?;
        let disk_mib = quantity_mib(&limits.disk).ok_or_else(|| {
            AlienError::new(ErrorData::SandboxLimitInvalid {
                resource_id: self.id.clone(),
                field: "disk".to_string(),
                value: limits.disk.clone(),
                reason: "AWS sizes a MicroVM's disk in whole MiB".to_string(),
            })
        })?;
        let cpu_millicores = millicores(&limits.cpu).ok_or_else(|| {
            AlienError::new(ErrorData::SandboxLimitInvalid {
                resource_id: self.id.clone(),
                field: "cpu".to_string(),
                value: limits.cpu.clone(),
                reason: "expected cores or millicores".to_string(),
            })
        })?;

        // Memory and disk choose the size; cpu is then checked rather than used to choose.
        // AWS couples cpu to memory at 2 GB per vCPU, so letting a low cpu ceiling select the
        // size too would quietly hand back a machine four times smaller than the memory ceiling
        // asked for, with nothing to indicate it.
        let sized = |tier: &&MicrovmTier| {
            tier.peak_memory_mib <= memory_mib && tier.max_disk_mib <= disk_mib
        };

        let tier = MICROVM_TIERS
            .iter()
            .rev()
            .find(sized)
            .copied()
            .ok_or_else(|| {
                AlienError::new(ErrorData::SandboxLimitInvalid {
                    resource_id: self.id.clone(),
                    field: "memory".to_string(),
                    value: limits.memory.clone(),
                    reason: format!(
                        "a Lambda MicroVM bursts to four times its baseline, so the smallest \
                         ceiling AWS can hold is 2Gi memory with 8Gi disk; '{}' memory and '{}' \
                         disk fit no size",
                        limits.memory, limits.disk
                    ),
                })
            })?;

        let required_millicores = i64::from(tier.peak_vcpu) * 1000;
        if cpu_millicores < required_millicores {
            return Err(AlienError::new(ErrorData::SandboxLimitInvalid {
                resource_id: self.id.clone(),
                field: "cpu".to_string(),
                value: limits.cpu.clone(),
                reason: format!(
                    "AWS allocates one vCPU per 2GB, so a MicroVM sized to a '{}' memory ceiling \
                     reaches {} vCPU; declare cpu '{}' or lower the memory ceiling",
                    limits.memory, tier.peak_vcpu, tier.peak_vcpu
                ),
            }));
        }

        Ok(tier)
    }

    /// The capability checks that do not depend on declared limits.
    fn validate_capabilities(
        &self,
        capabilities: &SandboxCapabilities,
        platform: Platform,
    ) -> Result<()> {
        if matches!(self.egress, SandboxEgress::AllowDomains { .. }) {
            capabilities.require(SandboxCapability::DomainEgressRules, platform)?;
        }

        // `allow` asks for no restriction, so a backend that ignores it fails loudly on the first
        // blocked connection. `deny` asks for one, and a backend that ignores it puts untrusted
        // code on the internet with nothing to notice — so only this direction is gated.
        // An empty list is not a restriction anyone wrote down: it renders as a deny-all wearing
        // an allowlist's label, which reads at a glance as the opposite of what it does.
        if let SandboxEgress::AllowDomains { domains } = &self.egress {
            if domains.is_empty() {
                return Err(AlienError::new(ErrorData::SandboxLimitInvalid {
                    resource_id: self.id.clone(),
                    field: "egress.domains".to_string(),
                    value: "[]".to_string(),
                    reason: "an allowlist naming no domain denies everything; declare \
                             egress: deny if that is what was meant"
                        .to_string(),
                }));
            }
        }

        if matches!(self.egress, SandboxEgress::Deny) {
            capabilities.require(SandboxCapability::EgressDeny, platform)?;
        }

        if !self.preview_ports.is_empty() {
            capabilities.require(SandboxCapability::Preview, platform)?;
        }

        if self.lifecycle.idle_pause_seconds.is_some() {
            capabilities.require(SandboxCapability::PauseResume, platform)?;
        }

        if self.lifecycle.max_lifetime_seconds.is_some() {
            capabilities.require(SandboxCapability::SandboxLifetime, platform)?;
        }

        Ok(())
    }
}

/// Ceilings applied when a declaration names none.
///
/// Modest on purpose: an undeclared sandbox is one whose author did not think about sizing, and
/// the safe reading of that is a small box rather than a generous one.
fn default_limits() -> SandboxLimits {
    SandboxLimits {
        cpu: "1".to_string(),
        memory: "2Gi".to_string(),
        disk: "8Gi".to_string(),
        max_processes: None,
    }
}

/// Validates a Kubernetes-style resource quantity such as `500m`, `2Gi` or `1`.
fn validate_quantity(resource_id: &str, field: &str, value: &str) -> Result<()> {
    let invalid = |reason: &str| {
        AlienError::new(ErrorData::SandboxLimitInvalid {
            resource_id: resource_id.to_string(),
            field: field.to_string(),
            value: value.to_string(),
            reason: reason.to_string(),
        })
    };

    let digits_end = value
        .find(|c: char| !c.is_ascii_digit() && c != '.')
        .unwrap_or(value.len());
    let (number, suffix) = value.split_at(digits_end);

    let parsed: f64 = number
        .parse()
        .map_err(|_| invalid("expected a number, optionally followed by a unit suffix"))?;

    if parsed <= 0.0 {
        return Err(invalid("must be greater than zero"));
    }

    const SUFFIXES: &[&str] = &["", "m", "k", "M", "G", "T", "Ki", "Mi", "Gi", "Ti"];
    if !SUFFIXES.contains(&suffix) {
        return Err(invalid(
            "unit must be one of m, k, M, G, T, Ki, Mi, Gi, Ti, or absent",
        ));
    }

    Ok(())
}

/// Splits a quantity into its number and unit suffix.
fn split_quantity(value: &str) -> Option<(f64, &str)> {
    let trimmed = value.trim();
    let digits_end = trimmed
        .find(|c: char| !c.is_ascii_digit() && c != '.')
        .unwrap_or(trimmed.len());
    let (number, suffix) = trimmed.split_at(digits_end);
    number.parse().ok().map(|number| (number, suffix))
}

/// A memory or disk quantity in whole MiB, rounded down.
///
/// Every suffix `validate_quantity` accepts is handled here. Reading only `Gi` and `Mi` and
/// falling back for the rest would turn a declared `4G` into a different size than the customer
/// asked for, which for a ceiling means a sandbox larger than its bound.
pub fn quantity_mib(value: &str) -> Option<i64> {
    let (number, suffix) = split_quantity(value)?;
    let bytes = match suffix {
        "" => number,
        "k" => number * 1e3,
        "M" => number * 1e6,
        "G" => number * 1e9,
        "T" => number * 1e12,
        "Ki" => number * 1024.0,
        "Mi" => number * 1024.0 * 1024.0,
        "Gi" => number * 1024.0 * 1024.0 * 1024.0,
        "Ti" => number * 1024.0 * 1024.0 * 1024.0 * 1024.0,
        // `m` is a millicore suffix; memory has no use for it.
        _ => return None,
    };
    Some((bytes / (1024.0 * 1024.0)) as i64)
}

/// A CPU quantity in millicores.
pub fn millicores(value: &str) -> Option<i64> {
    let (number, suffix) = split_quantity(value)?;
    match suffix {
        "" => Some((number * 1000.0) as i64),
        "m" => Some(number as i64),
        _ => None,
    }
}

/// Outputs generated by a successfully provisioned Sandbox parent.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(rename_all = "camelCase")]
pub struct SandboxOutputs {
    /// Name of the durable parent that sandboxes are created inside
    pub parent_name: String,
    /// Platform-specific identifier for the parent (image ARN, sandbox group id, namespace)
    #[serde(skip_serializing_if = "Option::is_none")]
    pub identifier: Option<String>,
    /// Data-plane endpoint sandboxes are created through, where the platform has one
    #[serde(skip_serializing_if = "Option::is_none")]
    pub endpoint: Option<String>,
}

impl ResourceOutputsDefinition for SandboxOutputs {
    fn get_resource_type(&self) -> ResourceType {
        Sandbox::RESOURCE_TYPE
    }

    fn as_any(&self) -> &dyn Any {
        self
    }

    fn box_clone(&self) -> Box<dyn ResourceOutputsDefinition> {
        Box::new(self.clone())
    }

    fn outputs_eq(&self, other: &dyn ResourceOutputsDefinition) -> bool {
        other.as_any().downcast_ref::<SandboxOutputs>() == Some(self)
    }

    fn to_json_value(&self) -> serde_json::Result<serde_json::Value> {
        serde_json::to_value(self)
    }
}

impl ResourceDefinition for Sandbox {
    fn get_resource_type(&self) -> ResourceType {
        Self::RESOURCE_TYPE
    }

    fn id(&self) -> &str {
        &self.id
    }

    fn get_dependencies(&self) -> Vec<ResourceRef> {
        Vec::new()
    }

    fn validate_update(&self, new_config: &dyn ResourceDefinition) -> Result<()> {
        let new_sandbox = new_config
            .as_any()
            .downcast_ref::<Sandbox>()
            .ok_or_else(|| {
                AlienError::new(ErrorData::UnexpectedResourceType {
                    resource_id: self.id.clone(),
                    expected: Self::RESOURCE_TYPE,
                    actual: new_config.get_resource_type(),
                })
            })?;

        if self.id != new_sandbox.id {
            return Err(AlienError::new(ErrorData::InvalidResourceUpdate {
                resource_id: self.id.clone(),
                reason: "the 'id' field is immutable".to_string(),
            }));
        }

        Ok(())
    }

    fn as_any(&self) -> &dyn Any {
        self
    }

    fn as_any_mut(&mut self) -> &mut dyn Any {
        self
    }

    fn box_clone(&self) -> Box<dyn ResourceDefinition> {
        Box::new(self.clone())
    }

    fn resource_eq(&self, other: &dyn ResourceDefinition) -> bool {
        other.as_any().downcast_ref::<Sandbox>() == Some(self)
    }

    fn to_json_value(&self) -> serde_json::Result<serde_json::Value> {
        serde_json::to_value(self)
    }
}

/// The one token a sandbox bundle URI may carry, replaced with the deploying region.
///
/// AWS builds a MicroVM image only from a bucket in the image's own region, so a vendor
/// publishing to every supported region needs one stored URI that resolves per region.
pub const BUNDLE_REGION_TOKEN: &str = "{region}";

/// A bundle URI split around its region token, or carried whole when it has none.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum BundleUri<'a> {
    /// No token: emitted exactly as it is today.
    Literal(&'a str),
    /// The text either side of the token, for an emitter to rejoin around its own region
    /// expression.
    Regional { before: &'a str, after: &'a str },
}

/// The prefix a rebuild's new key still sits under: everything above the file name and the
/// version segment beneath it. `None` when nothing sits there, meaning no prefix can be granted
/// without also granting objects a rebuild never reads. Shared so both emitters agree on it.
pub fn stable_bundle_key_prefix(key: &str) -> Option<&str> {
    let (above_file, _) = key.rsplit_once('/')?;
    let (above_version, _) = above_file.rsplit_once('/')?;
    Some(above_version)
}

/// Reads a sandbox bundle URI, refusing anything an image build would only reject later.
///
/// The token is accepted in the bucket alone. A key-position token would name an object that does
/// not exist, and any other brace is a typo that would otherwise reach S3 verbatim and fail ~160s
/// into the build — which is the failure this whole check exists to move to plan time.
pub fn parse_bundle_uri(uri: &str) -> std::result::Result<BundleUri<'_>, String> {
    let path = uri
        .strip_prefix("s3://")
        .ok_or_else(|| format!("'{uri}' is not an s3:// URI"))?;
    let (bucket, key) = path
        .split_once('/')
        .ok_or_else(|| format!("'{uri}' names a bucket with no object key"))?;

    // Both emitters interpolate this path into the build role's resource ARN, where `*` and `?`
    // are IAM wildcards rather than literal characters. S3 accepts them in a key, so a bundle
    // published under one would silently widen the grant past the bundle it names.
    if path.contains('*') || path.contains('?') {
        return Err(format!(
            "'{uri}' carries an IAM wildcard; the bundle's path is interpolated into the build \
             role's grant, so '*' and '?' would widen it past the bundle"
        ));
    }

    if key.contains('{') || key.contains('}') {
        return Err(format!(
            "'{uri}' places a token in the object key; {BUNDLE_REGION_TOKEN} is accepted in the \
             bucket name alone"
        ));
    }

    let Some((before, after)) = bucket.split_once(BUNDLE_REGION_TOKEN) else {
        if bucket.contains('{') || bucket.contains('}') {
            return Err(format!(
                "'{uri}' carries a token this build does not know; {BUNDLE_REGION_TOKEN} is the \
                 only one"
            ));
        }
        return Ok(BundleUri::Literal(uri));
    };

    if after.contains(BUNDLE_REGION_TOKEN) {
        return Err(format!("'{uri}' repeats {BUNDLE_REGION_TOKEN}"));
    }
    if before.contains('{') || before.contains('}') || after.contains('{') || after.contains('}') {
        return Err(format!(
            "'{uri}' carries a token this build does not know; {BUNDLE_REGION_TOKEN} is the only one"
        ));
    }

    Ok(BundleUri::Regional {
        before: &uri[.."s3://".len() + before.len()],
        after: &uri["s3://".len() + before.len() + BUNDLE_REGION_TOKEN.len()..],
    })
}

/// Where a private ECR image's region comes from.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum EcrImageRegion<'a> {
    /// A region named in the host.
    Literal(&'a str),
    /// [`BUNDLE_REGION_TOKEN`] in the host: the region the deployment renders.
    Deployment,
}

/// The ECR repository a private image reference is pulled from.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct EcrImageRepository<'a> {
    pub account_id: &'a str,
    pub region: EcrImageRegion<'a>,
    /// The repository name, which may carry `/`; never a tag or digest.
    pub repository: &'a str,
}

impl EcrImageRepository<'_> {
    /// The repository's ARN, with `region` standing in for a region the host leaves to the
    /// deployment. The partition is the deployment's: a GovCloud host ends `.amazonaws.com` too.
    pub fn arn(&self, partition: &str, region: &str) -> String {
        let region = match self.region {
            EcrImageRegion::Literal(region) => region,
            EcrImageRegion::Deployment => region,
        };
        format!(
            "arn:{partition}:ecr:{region}:{}:repository/{}",
            self.account_id, self.repository
        )
    }
}

/// Reads `privateBaseImage` as the one repository a build role may pull from.
///
/// The name is interpolated into an IAM ARN, so it is held to ECR's own repository grammar: that
/// refuses `*` and `?`, which would widen the grant, and `$` and braces, which a CloudFormation
/// `Sub` or a Terraform template would read as an expression.
pub fn parse_ecr_image_repository(
    image: &str,
) -> std::result::Result<EcrImageRepository<'_>, String> {
    let refuse = |reason: &str| format!("privateBaseImage '{image}' {reason}");
    let (host, path) = image
        .split_once('/')
        .ok_or_else(|| refuse("names no repository"))?;
    let (account_id, rest) = host.split_once(".dkr.ecr.").ok_or_else(|| {
        refuse("is not served by a private ECR registry (<account>.dkr.ecr.<region>.amazonaws.com)")
    })?;
    // `.com.cn` first: a China host ends with the shorter suffix too.
    let region = rest
        .strip_suffix(".amazonaws.com.cn")
        .or_else(|| rest.strip_suffix(".amazonaws.com"))
        .ok_or_else(|| refuse("is not served by a private ECR registry (<account>.dkr.ecr.<region>.amazonaws.com)"))?;
    if account_id.len() != 12 || !account_id.bytes().all(|b| b.is_ascii_digit()) {
        return Err(refuse("names no 12-digit account in its registry host"));
    }
    let region = if region == BUNDLE_REGION_TOKEN {
        EcrImageRegion::Deployment
    } else if !region.is_empty()
        && region
            .bytes()
            .all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'-')
    {
        EcrImageRegion::Literal(region)
    } else {
        return Err(refuse(&format!(
            "names no region in its registry host; give one or {BUNDLE_REGION_TOKEN}"
        )));
    };

    let repository = match path.split_once('@') {
        Some((repository, _digest)) => repository,
        None => match path.rsplit_once('/') {
            Some((parent, last)) => match last.split_once(':') {
                Some((name, _tag)) => &path[..parent.len() + 1 + name.len()],
                None => path,
            },
            None => path.split_once(':').map_or(path, |(name, _tag)| name),
        },
    };
    let valid_segment = |segment: &str| {
        !segment.is_empty()
            && segment.bytes().all(|b| {
                b.is_ascii_lowercase() || b.is_ascii_digit() || matches!(b, b'.' | b'_' | b'-')
            })
    };
    if !repository.split('/').all(valid_segment) {
        return Err(refuse(
            "names a repository outside ECR's grammar (lowercase letters, digits, '.', '_', '-', and '/' between them)",
        ));
    }
    Ok(EcrImageRepository {
        account_id,
        region,
        repository,
    })
}

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

    #[test]
    fn private_database_setup_accepts_the_default_network() {
        for lifecycle in [
            crate::ResourceLifecycle::Frozen,
            crate::ResourceLifecycle::Live,
        ] {
            let stack = crate::Stack::new("database".to_string())
                .add(
                    crate::Postgres::new("metadata".to_string()).build(),
                    lifecycle,
                )
                .build();
            assert!(!restricts_network_mode(&stack, false));
            assert!(!restricts_network_mode(&stack, true));
        }
        assert!(!restricts_network_mode(
            &crate::Stack::new("empty".to_string()).build(),
            false,
        ));
    }

    #[test]
    fn a_private_base_image_names_one_repository() {
        let deployment = EcrImageRegion::Deployment;
        let literal = EcrImageRegion::Literal;
        let accepted = [
            (
                "123456789012.dkr.ecr.us-east-1.amazonaws.com/base:1.0",
                literal("us-east-1"),
                "base",
            ),
            (
                "123456789012.dkr.ecr.us-east-1.amazonaws.com/team/agents/base:1.0",
                literal("us-east-1"),
                "team/agents/base",
            ),
            (
                "123456789012.dkr.ecr.{region}.amazonaws.com/team/base@sha256:abc123",
                deployment,
                "team/base",
            ),
            (
                "123456789012.dkr.ecr.cn-north-1.amazonaws.com.cn/base",
                literal("cn-north-1"),
                "base",
            ),
            (
                "123456789012.dkr.ecr.us-gov-west-1.amazonaws.com/my.base_image-x",
                literal("us-gov-west-1"),
                "my.base_image-x",
            ),
        ];
        for (image, region, repository) in accepted {
            assert_eq!(
                parse_ecr_image_repository(image),
                Ok(EcrImageRepository {
                    account_id: "123456789012",
                    region,
                    repository,
                }),
                "{image}"
            );
        }

        for refused in [
            "public.ecr.aws/docker/library/alpine:3.20",
            "docker.io/library/alpine:3.20",
            "https://123456789012.dkr.ecr.us-east-1.amazonaws.com/base:1.0",
            "123456789012.dkr.ecr.us-east-1.amazonaws.com",
            "123456789012.dkr.ecr.us-east-1.amazonaws.com/",
            "12345.dkr.ecr.us-east-1.amazonaws.com/base",
            "123456789012.dkr.ecr..amazonaws.com/base",
            "123456789012.dkr.ecr.{account}.amazonaws.com/base",
            "123456789012.dkr.ecr.us-east-1.amazonaws.com/*",
            "123456789012.dkr.ecr.us-east-1.amazonaws.com/ba?e",
            "123456789012.dkr.ecr.us-east-1.amazonaws.com/${AWS::AccountId}",
            "123456789012.dkr.ecr.us-east-1.amazonaws.com/{region}/base",
            "123456789012.dkr.ecr.us-east-1.amazonaws.com/Base:1.0",
            "123456789012.dkr.ecr.us-east-1.amazonaws.com/team//base",
        ] {
            assert!(
                parse_ecr_image_repository(refused).is_err(),
                "{refused} must be refused"
            );
        }
    }

    #[test]
    fn a_repository_arn_takes_the_deployment_region_only_where_the_host_leaves_it() {
        let regional =
            parse_ecr_image_repository("123456789012.dkr.ecr.{region}.amazonaws.com/team/base:1")
                .expect("parses");
        let pinned =
            parse_ecr_image_repository("123456789012.dkr.ecr.eu-west-1.amazonaws.com/team/base:1")
                .expect("parses");
        assert_eq!(
            regional.arn("aws-us-gov", "us-gov-west-1"),
            "arn:aws-us-gov:ecr:us-gov-west-1:123456789012:repository/team/base"
        );
        assert_eq!(
            pinned.arn("aws", "us-east-1"),
            "arn:aws:ecr:eu-west-1:123456789012:repository/team/base"
        );
    }

    /// A wildcard reaching the grant would widen it past the bundle, and it widens the Frozen
    /// object grant as readily as the Live prefix — both interpolate the path into the ARN.
    #[test]
    fn a_uri_carrying_an_iam_wildcard_is_refused() {
        for uri in [
            "s3://acme/team-*/v1/bundle.zip",
            "s3://acme/sandbox-bundle/f00d/bundle?.zip",
            "s3://acme-*/sandbox-bundle/f00d/bundle.zip",
        ] {
            let error = parse_bundle_uri(uri).expect_err("a wildcard must be refused");
            assert!(error.contains("IAM wildcard"), "for {uri}: {error}");
        }

        parse_bundle_uri("s3://acme/sandbox-bundle/f00d/bundle.zip")
            .expect("an ordinary key still parses");
    }

    /// The rule both package formats grant by, pinned here rather than in either. The
    /// near-misses the cases separate: the key's first segment grants objects a rebuild never
    /// reads, and the object's own directory pins the version segment that moves.
    #[test]
    fn a_grantable_prefix_stops_above_the_segment_that_moves() {
        assert_eq!(
            stable_bundle_key_prefix("sandbox-bundle/f00dcafe/bundle.zip"),
            Some("sandbox-bundle")
        );
        assert_eq!(
            stable_bundle_key_prefix("artifacts/team-a/sandbox/f00dcafe/bundle.zip"),
            Some("artifacts/team-a/sandbox"),
            "a deeper key narrows the prefix, it never widens to the first segment"
        );

        // Nothing sits above the version segment, so no prefix a moved bundle stays inside
        // exists. Emitting the object grant instead installs a role that denies the next rebuild.
        assert_eq!(stable_bundle_key_prefix("agents/bundle.zip"), None);
        assert_eq!(stable_bundle_key_prefix("bundle.zip"), None);
    }

    fn sandbox_with(egress: SandboxEgress, preview_ports: Vec<u16>) -> Sandbox {
        Sandbox::new("agent-sbx".to_string())
            .code(SandboxCode::Image {
                image: "ubuntu".to_string(),
            })
            .limits(SandboxLimits {
                cpu: "1".to_string(),
                memory: "2Gi".to_string(),
                disk: "20Gi".to_string(),
                max_processes: None,
            })
            .egress(egress)
            .lifecycle(SandboxLifecyclePolicy {
                max_lifetime_seconds: None,
                idle_pause_seconds: None,
            })
            .preview_ports(preview_ports)
            .build()
    }

    /// A URI with no token must come back whole, because every bundle configured today has none
    /// and emitting one differently would change every existing customer's template.
    #[test]
    fn a_uri_without_a_token_is_carried_whole() {
        assert_eq!(
            parse_bundle_uri("s3://acme-artifacts-us-east-2/agents/bundle.zip"),
            Ok(BundleUri::Literal(
                "s3://acme-artifacts-us-east-2/agents/bundle.zip"
            ))
        );
    }

    /// The split has to rejoin to the original with the region in place, or an emitter builds a
    /// URI that is subtly not the one the vendor configured.
    #[test]
    fn a_regional_uri_splits_either_side_of_the_token() {
        let BundleUri::Regional { before, after } =
            parse_bundle_uri("s3://acme-artifacts-{region}/agents/bundle.zip")
                .expect("the token is accepted in the bucket")
        else {
            panic!("a bucket-position token must split");
        };

        assert_eq!(before, "s3://acme-artifacts-");
        assert_eq!(after, "/agents/bundle.zip");
        assert_eq!(
            format!("{before}us-east-2{after}"),
            "s3://acme-artifacts-us-east-2/agents/bundle.zip",
            "the halves must rejoin to the URI the vendor meant"
        );
    }

    /// Each of these reaches S3 verbatim and dies ~160s into an image build if it is not refused
    /// here, which is the whole reason this runs at plan time.
    #[test]
    fn a_token_this_build_cannot_resolve_is_refused() {
        for uri in [
            "s3://acme-artifacts-{regio}/bundle.zip",
            "s3://acme-artifacts/{region}/bundle.zip",
            "s3://acme-artifacts-{region}-{region}/bundle.zip",
            "s3://acme-artifacts/bundle-{version}.zip",
            "s3://acme}-artifacts-{region}/bundle.zip",
            "s3://acme{-artifacts-{region}/bundle.zip",
        ] {
            assert!(
                parse_bundle_uri(uri).is_err(),
                "'{uri}' must be refused before it can reach an image build"
            );
        }
    }

    #[test]
    fn resource_type_is_stable() {
        assert_eq!(Sandbox::RESOURCE_TYPE.as_ref(), "sandbox");
    }

    #[test]
    fn capability_sets_are_per_platform() {
        let gcp = SandboxCapabilities::for_platform(Platform::Gcp).expect("gcp is supported");
        assert!(
            gcp.reconnect,
            "generation from the container boot id makes a sandbox reachable across processes"
        );
        assert!(!gcp.preview);
        assert!(gcp.enforced_limits);

        let azure = SandboxCapabilities::for_platform(Platform::Azure).expect("azure is supported");
        assert!(azure.files, "every backend moves files");
        assert!(gcp.files);
        // Azure is the only backend whose egress policy matches on host pattern, and the only
        // one where `deny` and a hostname list are the same object.
        assert!(azure.domain_egress_rules);
        assert!(azure.egress_deny);
        // The data plane honours a continuous cpu/memory/disk surface and refuses anything
        // outside `n×250m`, with the rule in the message.
        assert!(azure.enforced_limits);
        assert!(azure.pause_resume);
        // Both stay false for reasons that are not "unbuilt": a snapshot id has nothing to
        // consume it on any backend, and an Azure port's auth is anonymous or a human allowlist,
        // neither of which is a port-scoped credential.
        assert!(!azure.snapshot);
        assert!(!azure.preview);

        let aws = SandboxCapabilities::for_platform(Platform::Aws).expect("aws is supported");
        assert!(!aws.snapshot, "AWS has no user-callable sandbox snapshot");
        assert!(aws.pause_resume);

        let k8s =
            SandboxCapabilities::for_platform(Platform::Kubernetes).expect("k8s is supported");
        assert!(
            !k8s.preview,
            "the sandbox-scoped ingress gateway does not exist yet"
        );
    }

    /// Whether the process supervising a command is a separate identity from the command.
    ///
    /// Values are measured, not inferred. AWS: the agent runs as uid 0 with
    /// `CapEff: 00000000a80425fb` and `setuid`s the command to uid 60000, so the two differ.
    /// Kubernetes: the sandbox pod pins `run_as_user: 65534` on both pod and container with
    /// `capabilities.drop: [ALL]` and `allow_privilege_escalation: false`, so no uid split is
    /// possible (`kubernetes_spec.rs`). Local: `docker exec` runs as the workload uid while the
    /// manager supervises from the host. Azure and Agent Platform have no in-sandbox supervisor.
    #[test]
    fn supervisor_isolation_is_per_platform() {
        let value = |platform| {
            SandboxCapabilities::for_platform(platform)
                .expect("supported")
                .supervisor_isolation
        };

        assert!(
            value(Platform::Aws),
            "root agent setuids the command to 60000"
        );
        assert!(
            value(Platform::Local),
            "the supervisor is on the host, outside the container"
        );
        assert!(
            !value(Platform::Kubernetes),
            "a single pinned uid cannot be split"
        );
        assert!(!value(Platform::Azure), "no Alien process runs the command");
        assert!(
            !value(Platform::Gcp),
            "no separate supervisor identity runs the command"
        );
    }

    /// The point of the field: AWS and GCP report the *same* `supervisor_pid_namespace` (neither
    /// has `CAP_SYS_ADMIN`), so that axis alone reads them as equivalent. They are not — AWS
    /// separates the command's identity from the supervisor's and Agent Platform does not.
    #[test]
    fn supervisor_isolation_separates_aws_from_a_subprocess_backend() {
        let aws = SandboxCapabilities::for_platform(Platform::Aws).expect("aws is supported");
        let gcp = SandboxCapabilities::for_platform(Platform::Gcp).expect("gcp is supported");

        assert_eq!(
            aws.supervisor_pid_namespace, gcp.supervisor_pid_namespace,
            "the older axis cannot tell them apart"
        );
        assert!(
            aws.supervisor_isolation,
            "AWS setuids the command off the supervisor"
        );
        assert!(
            !gcp.supervisor_isolation,
            "the command runs under no separate supervisor identity"
        );
    }

    /// The Agent Platform row, each value against the behaviour it was measured from. `reconnect`
    /// is the tripwire: it is `true` only because `generation` is derived from the container boot
    /// id read through the agent's health op, so a caller detects a replaced container instead of
    /// reconnecting to a blank one. It is also the body of the `Platform::Gcp` arm, asserted below.
    #[test]
    fn gcp_agent_platform_row_matches_measured_backend() {
        let row = SandboxCapabilities::gcp_agent_platform();

        assert!(row.files, "agent file ops move over the sandbox envelope");
        assert!(
            row.reconnect,
            "generation is derived from the container boot id, so a sandbox is reachable across \
             processes"
        );
        assert!(
            !row.preview,
            "the only ingress is :execute; no port-scoped capability"
        );
        assert!(
            row.pause_resume,
            ":pause and :resume preserve the container"
        );
        assert!(
            !row.snapshot,
            "the create path never sends a snapshot, so none is reachable through the trait"
        );
        assert!(
            !row.domain_egress_rules,
            "VPC and DNS peering is not a hostname allowlist"
        );
        assert!(
            row.egress_deny,
            "a declared deny blocks both egress and DNS"
        );
        assert!(
            row.enforced_limits,
            "ceilings are enforced, by terminating the sandbox on breach"
        );
        assert!(!row.process_limit, "no process-count ceiling is observed");
        assert!(row.sandbox_lifetime, "ttl maps to a sandbox expireTime");
        assert!(!row.supervisor_pid_namespace, "no PID-namespace isolation");
        assert!(
            !row.supervisor_isolation,
            "the command is not run under a separate supervisor identity"
        );

        // Agent Platform is the registered GCP backend, so the arm returns exactly this row.
        let live = SandboxCapabilities::for_platform(Platform::Gcp).expect("gcp is supported");
        assert_eq!(
            live, row,
            "the Platform::Gcp arm is the Agent Platform capability row"
        );
    }

    #[test]
    fn platforms_without_a_backend_are_an_error_not_an_empty_set() {
        let error = SandboxCapabilities::for_platform(Platform::Machines)
            .expect_err("Machines has no sandbox backend");
        assert_eq!(error.code, "SANDBOX_PLATFORM_UNSUPPORTED");
    }

    #[test]
    fn unsupported_capability_names_platform_and_capability() {
        let capabilities = SandboxCapabilities::for_platform(Platform::Gcp).expect("supported");
        let error = capabilities
            .require(SandboxCapability::Preview, Platform::Gcp)
            .expect_err("GCP has no preview");

        assert_eq!(error.code, "SANDBOX_CAPABILITY_UNSUPPORTED");
        let rendered = error.to_string();
        assert!(
            rendered.contains("preview"),
            "names the capability: {rendered}"
        );
        assert!(rendered.contains("gcp"), "names the platform: {rendered}");
    }

    /// Azure matches on hostname; AWS and Kubernetes match CIDRs, and Local and GCP have a
    /// switch rather than a filter. Accepting a hostname list on those four would leave a stack
    /// reading as restricted while the sandbox reaches the whole internet.
    #[test]
    fn a_hostname_allowlist_is_refused_everywhere_it_would_be_approximated() {
        let sandbox = sandbox_with(
            SandboxEgress::AllowDomains {
                domains: vec!["example.com".to_string()],
            },
            vec![],
        );

        for platform in [
            Platform::Aws,
            Platform::Gcp,
            Platform::Kubernetes,
            Platform::Local,
        ] {
            let error = sandbox
                .validate_for_platform(platform)
                .expect_err("only Azure expresses a hostname allowlist");
            assert_eq!(
                error.code, "SANDBOX_CAPABILITY_UNSUPPORTED",
                "on {platform:?}"
            );
        }

        assert!(
            SandboxCapabilities::for_platform(Platform::Azure)
                .expect("supported")
                .domain_egress_rules,
            "Azure's egress policy matches on host pattern"
        );
    }

    /// `deny` is the declaration that carries a security promise, so a backend that cannot keep
    /// it has to refuse rather than accept it and run the code with open egress.
    #[test]
    fn a_denied_egress_is_refused_where_it_would_not_be_enforced() {
        let sandbox = sandbox_with(SandboxEgress::Deny, vec![]);

        // GCP is asserted at the capability rather than through validation: this sandbox declares
        // ceilings GCP cannot enforce, so it is refused for a reason unrelated to egress.
        assert!(
            SandboxCapabilities::for_platform(Platform::Gcp)
                .expect("supported")
                .egress_deny
        );

        for platform in [Platform::Aws, Platform::Kubernetes, Platform::Local] {
            sandbox
                .validate_for_platform(platform)
                .expect("deny is enforced here");
        }

        // Declares no ceilings, which Azure refuses for its own reason, so this isolates egress.
        let egress_only = Sandbox::new("sbx".to_string())
            .code(SandboxCode::Image {
                image: "alpine".to_string(),
            })
            .egress(SandboxEgress::Deny)
            .lifecycle(SandboxLifecyclePolicy {
                max_lifetime_seconds: None,
                idle_pause_seconds: None,
            })
            .build();

        egress_only
            .validate_for_platform(Platform::Azure)
            .expect("Azure creates the sandbox under a Deny policy with full inspection");
    }

    /// A sandbox naming no ceilings is valid on every platform and still resolves to a concrete
    /// set. The rule Azure applies to ceilings that *are* declared is pinned separately, by
    /// `azure_sizes_follow_the_rule_the_data_plane_states`.
    #[test]
    fn a_sandbox_declaring_no_ceilings_takes_the_platforms_own() {
        let undeclared = Sandbox::new("sbx".to_string())
            .code(SandboxCode::Image {
                image: "alpine".to_string(),
            })
            .egress(SandboxEgress::Deny)
            .lifecycle(SandboxLifecyclePolicy {
                max_lifetime_seconds: None,
                idle_pause_seconds: None,
            })
            .build();

        undeclared
            .validate_for_platform(Platform::Azure)
            .expect("a sandbox naming no ceilings takes the platform's own");

        // A backend still gets a concrete set, so nothing downstream has to invent one.
        assert_eq!(undeclared.resolved_limits().cpu, "1");
    }

    /// Pins Azure's own sizing rule: `250m`, `1500m` and `4000m` are valid, `32000m` and `333m`
    /// are refused. Checked at plan time so a bad value is a declaration to fix, not a package
    /// that renders without error and dies at the first sandbox.
    #[test]
    fn azure_sizes_follow_the_rule_the_data_plane_states() {
        let sized = |cpu: &str, memory: &str, disk: &str| {
            let mut sandbox = sandbox_with(SandboxEgress::Deny, vec![]);
            let limits = sandbox
                .limits
                .as_mut()
                .expect("the fixture declares limits");
            limits.cpu = cpu.to_string();
            limits.memory = memory.to_string();
            limits.disk = disk.to_string();
            sandbox.validate_for_platform(Platform::Azure)
        };

        sized("250m", "512Mi", "5120Mi").expect("the smallest step the data plane accepts");
        sized("4000m", "8192Mi", "40960Mi").expect("cpu, memory and disk are all honoured");
        sized("16000m", "32Gi", "320Gi").expect("the top of the range");

        // Same off-step case `azure_sandbox_limits` checks the multiple for, not just the range.
        let off_step = sized("333m", "512Mi", "5120Mi").expect_err("333m is not a step of 250m");
        assert_eq!(off_step.code, "SANDBOX_LIMIT_INVALID", "{off_step}");
        assert!(off_step.to_string().contains("cpu"), "{off_step}");

        let too_big = sized("32000m", "64Gi", "640Gi").expect_err("32 cores is over the ceiling");
        assert_eq!(too_big.code, "SANDBOX_LIMIT_INVALID", "{too_big}");

        // Both ceilings are derived from the cpu, so the same memory passes at one size and fails
        // at another - which is what makes them worth checking rather than bounding absolutely.
        sized("1000m", "2Gi", "20Gi").expect("2Gi is exactly one core's worth");
        let over_memory = sized("250m", "2Gi", "5120Mi").expect_err("2Gi needs a full core");
        assert_eq!(over_memory.code, "SANDBOX_LIMIT_INVALID", "{over_memory}");
        assert!(over_memory.to_string().contains("memory"), "{over_memory}");

        let over_disk = sized("250m", "512Mi", "20Gi").expect_err("20Gi needs a full core");
        assert!(over_disk.to_string().contains("disk"), "{over_disk}");
    }

    #[test]
    fn preview_ports_require_the_preview_capability() {
        let sandbox = sandbox_with(SandboxEgress::Deny, vec![8080]);

        sandbox
            .validate_for_platform(Platform::Aws)
            .expect("AWS mints a port-scoped JWE");

        let error = sandbox
            .validate_for_platform(Platform::Kubernetes)
            .expect_err("Kubernetes preview is deferred");
        assert_eq!(error.code, "SANDBOX_CAPABILITY_UNSUPPORTED");
    }

    /// A grant nothing reads is the silent no-op the capability contract exists to prevent, and
    /// here it is worse than useless: the reader would take it for a base image that needs
    /// authenticating while the platform pulls `code.image` itself.
    #[test]
    fn a_private_base_image_is_refused_off_aws() {
        let mut sandbox = sandbox_with(SandboxEgress::Allow, vec![]);
        sandbox.code = SandboxCode::Image {
            image: "s3://acme-artifacts/agents/bundle.zip".to_string(),
        };
        sandbox.private_base_image =
            Some("123456789012.dkr.ecr.{region}.amazonaws.com/acme:tag".to_string());

        sandbox
            .validate_for_platform(Platform::Aws)
            .expect("AWS builds its image from a bundle, so a base image sits behind code.image");

        for platform in [
            Platform::Gcp,
            Platform::Azure,
            Platform::Kubernetes,
            Platform::Local,
        ] {
            let error = sandbox
                .validate_for_platform(platform)
                .expect_err("a backend that builds no image must refuse a base image for one");
            assert_eq!(error.code, "SANDBOX_CAPABILITY_UNSUPPORTED");
            assert!(
                error.to_string().contains("privateBaseImage"),
                "the refusal must name the field the user declared: {error}"
            );
        }
    }

    #[test]
    fn gcp_accepts_a_sandbox_declaring_enforced_limits() {
        let sandbox = sandbox_with(SandboxEgress::Allow, vec![]);
        sandbox
            .validate_for_platform(Platform::Gcp)
            .expect("Agent Platform enforces declared ceilings, by terminating on breach");
    }

    #[test]
    fn invalid_quantities_are_rejected_with_the_offending_field() {
        let mut sandbox = sandbox_with(SandboxEgress::Deny, vec![]);
        sandbox
            .limits
            .as_mut()
            .expect("the fixture declares limits")
            .memory = "2Gb".to_string();

        let error = sandbox
            .validate_for_platform(Platform::Aws)
            .expect_err("Gb is not a valid suffix");
        assert_eq!(error.code, "SANDBOX_LIMIT_INVALID");
        assert!(error.to_string().contains("memory"));

        sandbox
            .limits
            .as_mut()
            .expect("the fixture declares limits")
            .memory = "2Gi".to_string();
        sandbox
            .limits
            .as_mut()
            .expect("the fixture declares limits")
            .cpu = "0".to_string();
        let error = sandbox
            .validate_for_platform(Platform::Aws)
            .expect_err("zero cpu is not a ceiling");
        assert_eq!(error.code, "SANDBOX_LIMIT_INVALID");
    }

    #[test]
    fn zero_max_processes_is_rejected() {
        let mut sandbox = sandbox_with(SandboxEgress::Deny, vec![]);
        sandbox
            .limits
            .as_mut()
            .expect("the fixture declares limits")
            .max_processes = Some(0);

        let error = sandbox
            .validate_for_platform(Platform::Local)
            .expect_err("a sandbox must be able to run at least one process");
        assert_eq!(error.code, "SANDBOX_LIMIT_INVALID");
        assert!(error.to_string().contains("maxProcesses"));
    }

    /// A process ceiling needs a container runtime. Kubernetes sets one per node rather than per
    /// pod, and neither MicroVMs nor Azure sandboxes expose one, so accepting the declaration
    /// anywhere else would mean carrying a bound nothing applies.
    #[test]
    fn a_process_ceiling_is_accepted_only_where_a_runtime_can_apply_it() {
        let mut sandbox = sandbox_with(SandboxEgress::Deny, vec![]);
        sandbox
            .limits
            .as_mut()
            .expect("the fixture declares limits")
            .max_processes = Some(256);

        sandbox
            .validate_for_platform(Platform::Local)
            .expect("Docker takes a pids limit");

        for platform in [Platform::Aws, Platform::Azure, Platform::Kubernetes] {
            let error = sandbox
                .validate_for_platform(platform)
                .expect_err("a process ceiling nothing applies must be refused");
            assert_eq!(error.code, "SANDBOX_CAPABILITY_UNSUPPORTED");
        }
    }

    /// Lambda rejects a run outside 1–28,800 rather than clamping it, so a value beyond that
    /// would pass planning, render into the package, and fail at the first sandbox. Kubernetes
    /// takes the same field with no such bound, so the check is AWS's alone.
    #[test]
    fn a_lifetime_aws_would_reject_is_refused_while_planning() {
        let mut sandbox = sandbox_with(SandboxEgress::Deny, vec![]);

        for seconds in [0, 28_801, 100_000] {
            sandbox.lifecycle.max_lifetime_seconds = Some(seconds);
            let error = sandbox
                .validate_for_platform(Platform::Aws)
                .expect_err("a lifetime outside what AWS runs is refused");
            assert_eq!(error.code, "SANDBOX_LIMIT_INVALID", "{seconds}s");

            // Kubernetes has no such ceiling, so the same declaration is fine there.
            sandbox
                .validate_for_platform(Platform::Kubernetes)
                .expect("the kubelet takes any activeDeadlineSeconds");
        }

        sandbox.lifecycle.max_lifetime_seconds = Some(28_800);
        sandbox
            .validate_for_platform(Platform::Aws)
            .expect("the ceiling itself is allowed");
    }

    /// An image reference Azure cannot honour is refused while planning, not at the first sandbox.
    ///
    /// `code.image`'s own documentation gives a tag and a registry path as examples — exactly
    /// what Azure cannot take, so this is the shape a customer is most likely to declare.
    #[test]
    fn an_image_azure_cannot_pull_is_refused_while_planning() {
        let mut sandbox = sandbox_with(SandboxEgress::Deny, vec![]);
        // Azure enforces no declared ceiling, so a sandbox carrying limits is refused before the
        // image is ever read.
        sandbox.limits = None;

        for image in [
            "ubuntu:24.04",
            "ghcr.io/myorg/sandbox:latest",
            "ubuntu@sha256:abc",
            "",
            "   ",
            "ubuntu latest",
            "ubuntu?x",
        ] {
            sandbox.code = SandboxCode::Image {
                image: image.to_string(),
            };
            let error = sandbox
                .validate_for_platform(Platform::Azure)
                .expect_err("an image Azure has nowhere to put is refused");
            assert_eq!(error.code, "SANDBOX_LIMIT_INVALID", "image '{image}'");

            // The same declaration is ordinary everywhere that pulls a reference.
            sandbox
                .validate_for_platform(Platform::Kubernetes)
                .expect("a registry reference is what every other backend takes");
        }

        for image in ["ubuntu", "ubuntu-22.04", "debian_slim"] {
            sandbox.code = SandboxCode::Image {
                image: image.to_string(),
            };
            sandbox
                .validate_for_platform(Platform::Azure)
                .unwrap_or_else(|error| panic!("'{image}' is a catalog name: {error}"));
        }

        // Surrounding space is trimmed rather than carried into the create body.
        sandbox.code = SandboxCode::Image {
            image: " ubuntu ".to_string(),
        };
        assert_eq!(
            sandbox
                .azure_catalog_image()
                .expect("a padded name is still a name"),
            "ubuntu"
        );
    }

    /// A deadline is accepted only where the platform itself terminates on it — the kubelet's
    /// `activeDeadlineSeconds` and Lambda's `maximumDurationInSeconds`. Everywhere else it would
    /// need a reaper that does not exist, so it is refused rather than accepted and dropped.
    #[test]
    fn a_sandbox_deadline_is_accepted_only_where_the_platform_applies_it() {
        let mut sandbox = sandbox_with(SandboxEgress::Deny, vec![]);
        sandbox.lifecycle.max_lifetime_seconds = Some(3600);

        sandbox
            .validate_for_platform(Platform::Kubernetes)
            .expect("the kubelet enforces activeDeadlineSeconds");
        sandbox
            .validate_for_platform(Platform::Aws)
            .expect("Lambda terminates the MicroVM at maximumDurationInSeconds");

        for platform in [Platform::Azure, Platform::Local] {
            let error = sandbox
                .validate_for_platform(platform)
                .expect_err("a deadline nothing applies must be refused");
            assert_eq!(error.code, "SANDBOX_CAPABILITY_UNSUPPORTED");
        }
    }

    /// A MicroVM bursts to four times its baseline with no way to opt out, so a ceiling is kept
    /// by choosing the size whose *peak* fits inside it. Sizing by baseline would hand back a
    /// sandbox that can reach four times what the customer declared.
    #[test]
    fn an_aws_size_is_chosen_so_its_peak_stays_inside_the_declared_ceiling() {
        let sandbox = sandbox_with(SandboxEgress::Deny, vec![]);
        let tier = sandbox
            .microvm_tier()
            .expect("2Gi/1cpu/20Gi is satisfiable");

        assert_eq!(
            tier.peak_memory_mib, 2048,
            "the peak is the declared ceiling"
        );
        assert_eq!(
            tier.baseline_memory_mib, 512,
            "which is a quarter of it as the baseline"
        );
        assert!(tier.max_disk_mib <= 20 * 1024);
    }

    /// AWS allocates one vCPU per 2GB, so a cpu ceiling below what the memory ceiling implies
    /// cannot be honoured together with it. Letting cpu choose the size instead would hand back a
    /// machine four times smaller than the memory asked for, with nothing to indicate it.
    #[test]
    fn a_cpu_ceiling_below_what_the_memory_implies_is_refused_not_quietly_downsized() {
        let mut sandbox = sandbox_with(SandboxEgress::Deny, vec![]);
        {
            let limits = sandbox
                .limits
                .as_mut()
                .expect("the fixture declares limits");
            limits.cpu = "1".to_string();
            limits.memory = "8Gi".to_string();
        }

        let error = sandbox
            .microvm_tier()
            .expect_err("1 cpu and 8Gi cannot both be ceilings on AWS");
        assert!(
            error.to_string().contains("4 vCPU"),
            "the refusal must say what the memory ceiling implies: {error}"
        );

        sandbox
            .limits
            .as_mut()
            .expect("the fixture declares limits")
            .cpu = "4".to_string();
        let tier = sandbox.microvm_tier().expect("4 cpu matches 8Gi");
        assert_eq!(tier.peak_memory_mib, 8192);
    }

    /// Below AWS's smallest peak there is no size that holds the ceiling, and rounding up to the
    /// nearest one would silently exceed it.
    #[test]
    fn an_aws_ceiling_smaller_than_any_size_is_refused_rather_than_rounded() {
        let mut sandbox = sandbox_with(SandboxEgress::Deny, vec![]);
        sandbox
            .limits
            .as_mut()
            .expect("the fixture declares limits")
            .memory = "1Gi".to_string();

        let error = sandbox
            .validate_for_platform(Platform::Aws)
            .expect_err("no MicroVM size peaks at or below 1Gi");
        assert_eq!(error.code, "SANDBOX_LIMIT_INVALID");
        assert!(
            error.to_string().contains("2Gi"),
            "the refusal must say what the smallest holdable ceiling is: {error}"
        );
    }

    /// `alien build` builds an AWS sandbox's base image, so source is a declaration there. On
    /// every other platform the image is pulled as declared, and an unbuilt source would schedule
    /// a pod that can never run, so the refusal still has to happen at plan time.
    #[test]
    fn source_code_is_refused_off_aws_rather_than_producing_a_broken_manifest() {
        let sandbox = Sandbox::new("agent".to_string())
            .code(SandboxCode::Source {
                src: "./sandbox".to_string(),
                toolchain: ToolchainConfig::Docker {
                    dockerfile: None,
                    build_args: None,
                    target: None,
                },
            })
            .egress(SandboxEgress::Deny)
            .lifecycle(SandboxLifecyclePolicy {
                max_lifetime_seconds: None,
                idle_pause_seconds: None,
            })
            .build();

        sandbox
            .validate_for_platform(Platform::Aws)
            .expect("an AWS sandbox base image is built by `alien build`");

        for platform in [
            Platform::Azure,
            Platform::Gcp,
            Platform::Kubernetes,
            Platform::Local,
        ] {
            let error = sandbox
                .validate_for_platform(platform)
                .expect_err("no backend builds a sandbox image from source here");
            assert_eq!(error.code, "SANDBOX_LIMIT_INVALID");
            assert!(
                error.to_string().contains("code.image"),
                "the refusal must say what to write instead: {error}"
            );
            assert!(
                error.to_string().contains(&platform.to_string()),
                "the refusal must name the platform that cannot build it: {error}"
            );
        }
    }

    /// `validate_quantity` accepts nine suffixes. Reading only `Gi` and `Mi` would size a
    /// declared `4G` as though it were `4Gi`, which for a ceiling means exceeding it.
    #[test]
    fn every_accepted_unit_converts_rather_than_falling_back() {
        assert_eq!(quantity_mib("2Gi"), Some(2048));
        assert_eq!(quantity_mib("512Mi"), Some(512));
        assert_eq!(quantity_mib("4G"), Some(3814));
        assert_eq!(quantity_mib("1Ti"), Some(1024 * 1024));
        assert_eq!(millicores("1"), Some(1000));
        assert_eq!(millicores("500m"), Some(500));
    }

    #[test]
    fn unknown_fields_are_rejected() {
        let json = r#"{
            "id": "sbx",
            "code": {"type": "image", "image": "ubuntu:24.04"},
            "limits": {"cpu": "1", "memory": "2Gi", "disk": "20Gi"},
            "egress": {"mode": "deny"},
            "lifecycle": {},
            "unexpected": true
        }"#;

        serde_json::from_str::<Sandbox>(json).expect_err("deny_unknown_fields must reject");
    }

    #[test]
    fn serialization_roundtrips() {
        let sandbox = sandbox_with(
            SandboxEgress::AllowDomains {
                domains: vec!["example.com".to_string()],
            },
            vec![8080, 9090],
        );

        let json = serde_json::to_string(&sandbox).expect("serializes");
        let restored: Sandbox = serde_json::from_str(&json).expect("deserializes");
        assert_eq!(sandbox, restored);
    }

    #[test]
    fn id_is_immutable_across_updates() {
        let original = sandbox_with(SandboxEgress::Deny, vec![]);
        let renamed = Sandbox::new("other".to_string())
            .code(SandboxCode::Image {
                image: "ubuntu".to_string(),
            })
            .limits(
                original
                    .limits
                    .clone()
                    .expect("the fixture declares limits"),
            )
            .egress(SandboxEgress::Deny)
            .lifecycle(SandboxLifecyclePolicy {
                max_lifetime_seconds: None,
                idle_pause_seconds: None,
            })
            .build();

        original
            .validate_update(&original.clone())
            .expect("an unchanged config is a valid update");
        original
            .validate_update(&renamed)
            .expect_err("renaming a sandbox is not an update");
    }

    /// Azure declares an idle-pause policy but not a wall-clock ceiling.
    ///
    /// The two travel together in `SandboxLifecyclePolicy` and are gated separately on purpose:
    /// Azure pauses on idle and has no maximum lifetime, so accepting one and refusing the
    /// other is the honest split rather than an inconsistency.
    #[test]
    fn azure_takes_an_idle_policy_and_still_refuses_a_lifetime_ceiling() {
        let with_policy = |lifecycle: SandboxLifecyclePolicy| {
            Sandbox::new("sbx".to_string())
                .code(SandboxCode::Image {
                    image: "ubuntu".to_string(),
                })
                .egress(SandboxEgress::Allow)
                .lifecycle(lifecycle)
                .build()
                .validate_for_platform(Platform::Azure)
        };

        with_policy(SandboxLifecyclePolicy {
            max_lifetime_seconds: None,
            idle_pause_seconds: Some(900),
        })
        .expect("Azure pauses a sandbox on idle");

        let error = with_policy(SandboxLifecyclePolicy {
            max_lifetime_seconds: Some(3600),
            idle_pause_seconds: None,
        })
        .expect_err("Azure has no wall-clock ceiling to enforce one with");
        assert_eq!(error.code, "SANDBOX_CAPABILITY_UNSUPPORTED");
        assert!(
            error.message.contains("sandboxLifetime"),
            "names the capability: {}",
            error.message
        );
    }

    /// An allowlist naming nothing is a deny-all wearing an allowlist's label.
    ///
    /// It renders as a `Deny` default with no rules — the shape the Azure provider adds a
    /// catch-all to avoid — and a reader scanning the declaration sees "allowDomains" and reads
    /// the opposite of what it does.
    #[test]
    fn an_allowlist_with_no_domains_is_refused() {
        let declared = |domains: Vec<String>| {
            Sandbox::new("sbx".to_string())
                .code(SandboxCode::Image {
                    image: "ubuntu".to_string(),
                })
                .egress(SandboxEgress::AllowDomains { domains })
                .lifecycle(SandboxLifecyclePolicy {
                    max_lifetime_seconds: None,
                    idle_pause_seconds: None,
                })
                .build()
                .validate_for_platform(Platform::Azure)
        };

        let error = declared(vec![]).expect_err("an empty allowlist must be refused");
        assert_eq!(error.code, "SANDBOX_LIMIT_INVALID");

        declared(vec!["api.example.com".to_string()])
            .expect("a named domain is what an allowlist is for");
    }

    /// The two expressible modes map to the boolean; a host list maps to nothing so the caller has
    /// to refuse rather than silently pick a side.
    #[test]
    fn internet_access_switch_maps_only_the_two_expressible_modes() {
        assert_eq!(SandboxEgress::Allow.internet_access_switch(), Some(true));
        assert_eq!(SandboxEgress::Deny.internet_access_switch(), Some(false));
        assert_eq!(
            SandboxEgress::AllowDomains {
                domains: vec!["api.example.com".to_string()]
            }
            .internet_access_switch(),
            None,
            "a host list has no boolean and must not be approximated"
        );
    }
}