alien_core/resources/

sandbox.rs

//! 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 sha2::{Digest, Sha256};
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`). AWS wants an
        /// `s3://` bundle; Azure takes a catalog name such as `ubuntu` or an amd64 registry
        /// image, told apart by syntax. Each refuses what it cannot take while planning.
        image: String,
        /// The container image a publisher built the `s3://` bundle in `image` from. An AWS
        /// deployment with a bundle Storage of its own renders its bundle from this image; one
        /// whose setup granted the publisher's store builds from `image`. Set by a publisher,
        /// never by authoring. Kept inside `code` so a controller that predates it ignores it.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        #[cfg_attr(feature = "openapi", schema(ignore))]
        bundled_from: Option<String>,
    },
    /// A Dockerfile `alien build` builds into the sandbox's base image.
    ///
    /// AWS, Azure and GCP, and docker only: the base image is a root filesystem, not a binary laid
    /// on one. `alien release` pushes it; on AWS the bundle layers the sandbox agent on afterwards,
    /// and on GCP the release layers it on and pushes the result.
    #[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;

/// Longest `activeDeadlineSeconds` the Kubernetes API server accepts on a pod. The field is an
/// int64, but pod validation caps it at the largest i32.
const KUBERNETES_MAX_LIFETIME_SECONDS: u32 = i32::MAX as u32;

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

pub use super::sandbox_policy::{is_public_ipv4, EgressPolicy};

/// Outbound policy. Only the new Internet value opts into gateway enforcement.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum SandboxEgress {
    None,
    Internet,
    /// Deprecated direct backend access, unchanged from released deployments.
    Allow,
    /// Deprecated backend hostname allowlist, unchanged from released deployments.
    AllowDomains {
        domains: Vec<String>,
    },
}

/// Deprecated object declarations preserve their released backend behavior.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(tag = "mode", rename_all = "camelCase", deny_unknown_fields)]
pub enum LegacySandboxEgress {
    /// Deprecated alias of "none".
    Deny,
    /// Deprecated direct backend access. Use "internet" to opt into the gateway,
    /// fixed outbound IP, connection logs and per-sandbox runtime policies.
    Allow,
    /// Deprecated backend hostname rules. Use "internet" and runtime policies
    /// to opt into gateway enforcement instead.
    AllowDomains { domains: Vec<String> },
}

#[derive(Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
enum SandboxEgressMode {
    None,
    Internet,
}

#[derive(Serialize, Deserialize)]
#[serde(untagged)]
enum SandboxEgressInput {
    Mode(SandboxEgressMode),
    Legacy(LegacySandboxEgress),
}

impl Serialize for SandboxEgress {
    fn serialize<S: serde::Serializer>(
        &self,
        serializer: S,
    ) -> std::result::Result<S::Ok, S::Error> {
        match self {
            Self::None => serializer.serialize_str("none"),
            Self::Internet => serializer.serialize_str("internet"),
            Self::Allow => LegacySandboxEgress::Allow.serialize(serializer),
            Self::AllowDomains { domains } => LegacySandboxEgress::AllowDomains {
                domains: domains.clone(),
            }
            .serialize(serializer),
        }
    }
}

impl<'de> Deserialize<'de> for SandboxEgress {
    fn deserialize<D: serde::Deserializer<'de>>(
        deserializer: D,
    ) -> std::result::Result<Self, D::Error> {
        Ok(match SandboxEgressInput::deserialize(deserializer)? {
            SandboxEgressInput::Mode(SandboxEgressMode::None)
            | SandboxEgressInput::Legacy(LegacySandboxEgress::Deny) => Self::None,
            SandboxEgressInput::Mode(SandboxEgressMode::Internet) => Self::Internet,
            SandboxEgressInput::Legacy(LegacySandboxEgress::Allow) => Self::Allow,
            SandboxEgressInput::Legacy(LegacySandboxEgress::AllowDomains { domains }) => {
                Self::AllowDomains { domains }
            }
        })
    }
}

#[cfg(feature = "openapi")]
impl utoipa::PartialSchema for SandboxEgress {
    fn schema() -> utoipa::openapi::RefOr<utoipa::openapi::schema::Schema> {
        use utoipa::openapi::schema::{ObjectBuilder, OneOfBuilder, Type};
        OneOfBuilder::new()
            .item(
                ObjectBuilder::new()
                    .schema_type(Type::String)
                    .enum_values(Some(["none", "internet"])),
            )
            .item(LegacySandboxEgress::schema())
            .into()
    }
}
#[cfg(feature = "openapi")]
impl utoipa::ToSchema for SandboxEgress {}

impl SandboxEgress {
    /// Deprecated spelling retained for Rust callers of the released API.
    #[allow(non_upper_case_globals)]
    #[deprecated(note = "use SandboxEgress::None")]
    pub const Deny: Self = Self::None;

    /// Backend switch; hostname rules must be handled by a backend matcher.
    pub fn internet_access_switch(&self) -> Option<bool> {
        match self {
            Self::None => Some(false),
            Self::Internet | Self::Allow => Some(true),
            Self::AllowDomains { .. } => None,
        }
    }
}

/// Per-session deadline metadata read by the external lifecycle reaper.
pub const SANDBOX_EXPIRY_LABEL: &str = "alienExpiresAtMs";

/// Azure gateway identities cannot renew without an in-guest agent.
pub const AZURE_GATEWAY_MAX_LIFETIME_SECONDS: u32 = 24 * 60 * 60;

/// 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.
    ///
    /// Stored state written before the rename calls this `idleSuspendSeconds`.
    #[serde(alias = "idleSuspendSeconds", 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,
    /// A running command takes input on stdin, can be signalled, and streams its output as it is
    /// written, so a caller can drive a REPL, a language server or a tool protocol over it.
    pub interactive_commands: 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,
                // The in-guest agent owns the process and its pipes, and the MicroVM proxy
                // streams a response while separate requests carry input and signals.
                interactive_commands: 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,
                // The pod runs the same agent, but this client does not drive its interactive
                // routes yet, so the capability is not claimed until it is proven here.
                interactive_commands: false,
                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,
                // The manager holds the exec's attached stdin for as long as the output stream
                // is open, so an interactive command needs nothing inside the sandbox.
                interactive_commands: true,
                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,
            // Not claimed: the agent-mode client does not drive the agent's interactive routes
            // yet, and the platform's own exec takes no stdin.
            interactive_commands: 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,
            // `:execute` is one request with one answer, capped at about 30 seconds. Input
            // cannot reach a command after its call has answered, so there is no stream to drive.
            interactive_commands: false,
            // No method mints a port-scoped ingress capability; the only ingress is `:execute`.
            preview: false,
            // `:resume` can return a fresh container while reporting success, so a pause does not
            // keep the sandbox's state.
            pause_resume: false,
            // 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::InteractiveCommands => self.interactive_commands,
            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,
    /// Writing to, signalling and streaming a running command
    InteractiveCommands,
    /// 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::InteractiveCommands => "interactiveCommands",
            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",
        }
    }
}

/// Opt-in supervisor-owned network enforcement. Caller requests cannot change this identity.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct SandboxPrivilegedSupervisor {
    /// Nonzero numeric uid and primary gid for every command, including the image entrypoint.
    pub command_uid: u32,
}

/// 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 be DNS-compatible: lowercase letters, digits and hyphens,
    /// starting and ending with a letter or digit, with no consecutive hyphens.
    #[builder(start_fn)]
    pub id: String,
    /// Runtime network links authored by preflight, not parent lifecycle dependencies.
    /// Sandbox has no public link builder; gateway reach is wired during planning.
    #[builder(field)]
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    #[cfg_attr(feature = "openapi", schema(ignore))]
    pub links: Vec<ResourceRef>,
    /// Where the sandbox's root filesystem comes from
    pub code: SandboxCode,
    /// Original image uid resolved from the built OCI archives. Absent on released artifacts
    /// that retain the legacy boot contract. This is build output, not an authoring option.
    #[builder(skip)]
    #[serde(default, skip_serializing_if = "Option::is_none")]
    #[cfg_attr(feature = "openapi", schema(ignore))]
    pub image_user: Option<u32>,
    /// 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,
    /// Default runtime policy for creates that omit one.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub default_policy: Option<EgressPolicy>,
    /// Have Alien's agent install the declared egress policy before running any image code.
    /// Unsupported backends refuse this at plan time.
    // Keep released declarations readable without offering a new authoring knob.
    #[builder(skip)]
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub privileged_supervisor: Option<SandboxPrivilegedSupervisor>,
    /// Sandbox lifetime ceiling and idle behaviour.
    ///
    /// Stored state written before the rename calls this `session`. A stack state or release
    /// that old must stay readable, otherwise its deployment can no longer be updated or deleted.
    #[serde(alias = "session")]
    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(|(_, entry)| {
        entry
            .config
            .downcast_ref::<Sandbox>()
            .is_some_and(|sandbox| crate::sandbox_egress::sandbox_needs_connector(stack, sandbox))
    })
}

impl Sandbox {
    pub(crate) fn setup_digest_value(&self) -> serde_json::Value {
        match &self.egress {
            SandboxEgress::None => serde_json::json!({"mode":"deny"}),
            value => serde_json::to_value(value).expect("egress serializes"),
        }
    }

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

    /// Released supervisor declarations retain their direct cloud route; only
    /// the new Internet declaration uses the deployment gateway.
    pub fn cloud_egress(&self) -> &SandboxEgress {
        if self.privileged_supervisor.is_some() && self.egress != SandboxEgress::Internet {
            &SandboxEgress::Allow
        } else {
            &self.egress
        }
    }

    /// Startup contract for the privileged agent; never supplied by an exec caller.
    pub fn supervisor_environment(&self) -> std::collections::BTreeMap<String, String> {
        let mut env = std::collections::BTreeMap::new();
        if let Some(supervisor) = &self.privileged_supervisor {
            env.insert(
                "ALIEN_SANDBOX_EXPECTED_EXEC_UID".to_string(),
                supervisor.command_uid.to_string(),
            );
            env.insert(
                "ALIEN_SANDBOX_EGRESS".to_string(),
                // A runtime default is mutable gateway policy, never a startup allowlist.
                if self.egress == SandboxEgress::None {
                    r#"{"mode":"deny"}"#.to_string()
                } else {
                    serde_json::to_string(&self.egress).expect("egress serializes")
                },
            );
        }
        env
    }

    /// 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<()> {
        if let Some(policy) = &self.default_policy {
            if self.egress != SandboxEgress::Internet {
                return Err(AlienError::new(ErrorData::SandboxEgressInvalid {
                    target: "defaultPolicy".into(),
                    reason: "defaultPolicy requires the new internet declaration".into(),
                }));
            }
            policy.validate()?;
        }
        if self.image_user == Some(0) && matches!(platform, Platform::Local | Platform::Kubernetes)
        {
            return Err(AlienError::new(ErrorData::SandboxCapabilityUnsupported {
                capability: "root image USER: this backend does not yet separate bounded root boot from non-root exec; use a non-root image USER".into(),
                platform: platform.to_string(),
            }));
        }
        let mut capabilities = SandboxCapabilities::for_platform(platform)?;
        if let Some(supervisor) = &self.privileged_supervisor {
            if platform != Platform::Aws {
                return Err(AlienError::new(ErrorData::SandboxCapabilityUnsupported {
                    capability: "privilegedSupervisor".to_string(),
                    platform: platform.to_string(),
                }));
            }
            if supervisor.command_uid == 0 || supervisor.command_uid == u32::MAX {
                return Err(AlienError::new(ErrorData::SandboxLimitInvalid {
                    resource_id: self.id.clone(),
                    field: "privilegedSupervisor.commandUid".to_string(),
                    value: supervisor.command_uid.to_string(),
                    reason: "must be a non-root Linux uid other than the invalid uid sentinel"
                        .to_string(),
                }));
            }
            if let SandboxEgress::AllowDomains { domains } = &self.egress {
                for domain in domains {
                    let hostname = domain.strip_suffix('.').unwrap_or(domain);
                    if hostname.len() > 253
                        || !hostname.split('.').all(|label| {
                            !label.is_empty()
                                && label.len() <= 63
                                && label
                                    .bytes()
                                    .all(|c| c.is_ascii_alphanumeric() || c == b'-')
                                && !label.starts_with('-')
                                && !label.ends_with('-')
                        })
                    {
                        return Err(AlienError::new(ErrorData::SandboxLimitInvalid {
                            resource_id: self.id.clone(), field: "egress.domains".to_string(), value: domain.clone(),
                            reason: "privileged supervisor allowlists require exact DNS hostnames; wildcards are unsupported".to_string(),
                        }));
                    }
                }
            }
            capabilities.domain_egress_rules = true;
        }

        // `alien build` builds an AWS, Azure or GCP 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 { .. })
            && !matches!(platform, Platform::Aws | Platform::Azure | Platform::Gcp)
        {
            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. A source
        // declaration has no image until `alien build` writes one.
        if platform == Platform::Azure && matches!(&self.code, SandboxCode::Image { .. }) {
            self.azure_image()?;
        }

        // Read before the limits, like the image, because a lifetime is declared whether or not
        // any limits are.
        if let Some(seconds) = self.lifecycle.max_lifetime_seconds {
            // First, so a platform with no deadline says so before any complaint about the value.
            capabilities.require(SandboxCapability::SandboxLifetime, platform)?;

            if seconds == 0 {
                return Err(AlienError::new(ErrorData::SandboxLimitInvalid {
                    resource_id: self.id.clone(),
                    field: "maxLifetimeSeconds".to_string(),
                    value: "0".to_string(),
                    reason: "a sandbox must be allowed to run for at least 1 second".to_string(),
                }));
            }

            // Each ceiling is the platform's own, and the platform rejects a value above it
            // rather than clamping, so it would pass planning, render into the package, and fail
            // at the first sandbox. The ceilings differ, which is why they sit under a platform
            // match rather than on the type.
            let ceiling = match platform {
                Platform::Aws => Some((AWS_MAX_LIFETIME_SECONDS, "AWS runs a MicroVM for")),
                Platform::Kubernetes => {
                    Some((KUBERNETES_MAX_LIFETIME_SECONDS, "Kubernetes runs a pod for"))
                }
                // The Agent Platform documents `ttl` without a valid range, so a value it refuses
                // still fails at the first sandbox rather than here.
                Platform::Gcp => None,
                // Never reached: the capability check above refuses Azure and Local, and
                // `for_platform` refuses Machines and Test at the top of this function.
                Platform::Azure | Platform::Local | Platform::Machines | Platform::Test => None,
            };
            if let Some((max, runs_for)) = ceiling {
                if seconds > max {
                    return Err(AlienError::new(ErrorData::SandboxLimitInvalid {
                        resource_id: self.id.clone(),
                        field: "maxLifetimeSeconds".to_string(),
                        value: seconds.to_string(),
                        reason: format!("{runs_for} between 1 and {max} seconds"),
                    }));
                }
            }
        }

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

        self.validate_capabilities(&capabilities, platform)
    }

    /// What Azure creates this sandbox from: a catalog name or a registry image, told apart by
    /// [`classify_azure_sandbox_image`]. Refused while planning, because a value the data plane
    /// rejects would otherwise surface at the first sandbox, long after the apply.
    pub fn azure_image(&self) -> Result<AzureSandboxImage<'_>> {
        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: "the sandbox still has unbuilt source code. Run 'alien build' first"
                    .to_string(),
            }));
        };

        classify_azure_sandbox_image(image).ok_or_else(|| {
            let image = image.trim();
            AlienError::new(ErrorData::SandboxLimitInvalid {
                resource_id: self.id.clone(),
                field: "code.image".to_string(),
                value: image.to_string(),
                reason: if image.is_empty() {
                    "a sandbox has to name an image".to_string()
                } else {
                    "Azure creates a sandbox from a catalog name such as 'ubuntu' or from a \
                     registry image such as 'docker.io/library/python:3.14-slim'"
                        .to_string()
                },
            })
        })
    }

    /// 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::None) {
            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 replace_after_failed_create_is_safe(&self) -> bool {
        // A create adopts an existing image with the same name, and deleting it deletes that image,
        // which this deployment may never have built.
        false
    }

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

/// What an Azure sandbox starts from.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AzureSandboxImage<'a> {
    /// A public catalog disk image, such as `ubuntu`, which the data plane names directly.
    Catalog(&'a str),
    /// A registry image, which the controller builds into a disk image in the sandbox's group
    /// before any sandbox can start from it.
    Registry(&'a str),
}

impl<'a> AzureSandboxImage<'a> {
    /// The declared value, trimmed.
    pub fn as_str(&self) -> &'a str {
        match self {
            Self::Catalog(value) | Self::Registry(value) => value,
        }
    }
}

/// Label key the controller writes on every disk image it builds, and the provider finds it by.
pub const AZURE_DISK_IMAGE_LABEL: &str = "alienImage";

/// Classifies a declared `code.image` for Azure, or `None` when it is neither kind. A bare
/// `[A-Za-z0-9._-]+` is checked first and always a catalog name; anything else must carry `/`,
/// `:` or `@` and parse as an OCI reference.
pub fn classify_azure_sandbox_image(image: &str) -> Option<AzureSandboxImage<'_>> {
    let image = image.trim();
    if image.is_empty() {
        return None;
    }
    if image
        .chars()
        .all(|c| c.is_ascii_alphanumeric() || matches!(c, '.' | '_' | '-'))
    {
        return Some(AzureSandboxImage::Catalog(image));
    }
    (image.contains(|c| matches!(c, '/' | ':' | '@')) && OCI_REFERENCE.is_match(image))
        .then_some(AzureSandboxImage::Registry(image))
}

/// The label value naming the disk image built from `reference`: a digest, since a label value
/// may not carry a reference's `/`, `:` and `@`.
pub fn azure_disk_image_label(reference: &str) -> String {
    let digest = format!("{:x}", Sha256::digest(reference.trim().as_bytes()));
    digest[..32].to_string()
}

/// The distribution reference grammar: `[host[:port]/]path[:tag][@digest]`.
static OCI_REFERENCE: std::sync::LazyLock<regex::Regex> = std::sync::LazyLock::new(|| {
    let domain_component = r"(?:[a-zA-Z0-9]|[a-zA-Z0-9][a-zA-Z0-9-]*[a-zA-Z0-9])";
    let domain = format!(r"{domain_component}(?:\.{domain_component})*(?::[0-9]+)?");
    let path_component = r"[a-z0-9]+(?:(?:[._]|__|-+)[a-z0-9]+)*";
    let name = format!(r"(?:{domain}/)?{path_component}(?:/{path_component})*");
    let tag = r"[A-Za-z0-9_][A-Za-z0-9_.-]{0,127}";
    let digest = r"[A-Za-z][A-Za-z0-9]*(?:[-_+.][A-Za-z][A-Za-z0-9]*)*:[0-9a-fA-F]{32,}";
    regex::Regex::new(&format!(r"^{name}(?::{tag})?(?:@{digest})?$"))
        .expect("the OCI reference grammar compiles")
});

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

    /// The vectors' `repositoryKey`: registry host and repository, never the tag or digest, so a
    /// new tag keeps the key. A reference the parser refuses stays whole.
    fn repository_key(image: &str) -> String {
        match parse_ecr_image_repository(image) {
            Ok(parsed) => {
                let host = image.split_once('/').map_or(image, |(host, _)| host);
                format!("{host}/{}", parsed.repository)
            }
            Err(_) => image.to_string(),
        }
    }

    #[test]
    fn a_private_base_image_names_one_repository() {
        let vectors: serde_json::Value = serde_json::from_str(include_str!(
            "../../tests/fixtures/ecr-image-repository-parity.json"
        ))
        .expect("the ECR repository vectors must be JSON");
        let field = |case: &serde_json::Value, name: &str| {
            case[name]
                .as_str()
                .unwrap_or_else(|| panic!("vector must carry {name}: {case}"))
                .to_string()
        };

        for case in vectors["accepted"].as_array().expect("accepted vectors") {
            let image = field(case, "image");
            let region = field(case, "region");
            let region = if region == BUNDLE_REGION_TOKEN {
                EcrImageRegion::Deployment
            } else {
                EcrImageRegion::Literal(&region)
            };
            assert_eq!(
                parse_ecr_image_repository(&image),
                Ok(EcrImageRepository {
                    account_id: &field(case, "accountId"),
                    region,
                    repository: &field(case, "repository"),
                }),
                "{image}"
            );
            assert_eq!(
                repository_key(&image),
                field(case, "repositoryKey"),
                "{image}"
            );
        }

        for case in vectors["refused"].as_array().expect("refused vectors") {
            let image = field(case, "image");
            assert!(
                parse_ecr_image_repository(&image).is_err(),
                "{image} must be refused"
            );
            assert_eq!(
                repository_key(&image),
                field(case, "repositoryKey"),
                "{image}"
            );
        }
    }

    #[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(),
                bundled_from: None,
            })
            .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,
            ":resume can return a fresh container, so a pause keeps no state"
        );
        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::None, 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(),
                bundled_from: None,
            })
            .egress(SandboxEgress::None)
            .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(),
                bundled_from: None,
            })
            .egress(SandboxEgress::None)
            .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::None, 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::None, 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(),
            bundled_from: None,
        };
        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::None, 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::None, 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::None, 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 above 28,800 seconds 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 a far higher ceiling, so 28,800 is AWS's alone. The lifetime is a
    /// lifecycle field, so a sandbox that declares no limits is held to it too.
    #[test]
    fn a_lifetime_aws_would_reject_is_refused_while_planning() {
        let with_limits = sandbox_with(SandboxEgress::None, vec![]);
        let mut without_limits = with_limits.clone();
        without_limits.limits = None;

        for (declared, mut sandbox) in [("with", with_limits), ("without", without_limits)] {
            for seconds in [28_801, 30_000, 100_000] {
                sandbox.lifecycle.max_lifetime_seconds = Some(seconds);
                let error = sandbox
                    .validate_for_platform(Platform::Aws)
                    .expect_err("a lifetime above what AWS runs is refused");
                assert_eq!(
                    error.code, "SANDBOX_LIMIT_INVALID",
                    "{seconds}s {declared} limits"
                );

                // Kubernetes' ceiling is far higher, so the same declaration is fine there.
                sandbox
                    .validate_for_platform(Platform::Kubernetes)
                    .expect("Kubernetes takes an activeDeadlineSeconds above 28,800");
            }

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

    /// A zero lifetime is a sandbox that may not run at all. The Kubernetes API server refuses an
    /// `activeDeadlineSeconds` below 1 when the pod is created, so planning refuses it on every
    /// platform, with or without limits declared.
    #[test]
    fn a_zero_lifetime_is_refused_while_planning() {
        let with_limits = sandbox_with(SandboxEgress::None, vec![]);
        let mut without_limits = with_limits.clone();
        without_limits.limits = None;

        for (declared, mut sandbox) in [("with", with_limits), ("without", without_limits)] {
            sandbox.lifecycle.max_lifetime_seconds = Some(0);
            for platform in [Platform::Aws, Platform::Kubernetes, Platform::Gcp] {
                let error = sandbox
                    .validate_for_platform(platform)
                    .expect_err("a zero lifetime is refused");
                assert_eq!(
                    error.code, "SANDBOX_LIMIT_INVALID",
                    "{platform} {declared} limits"
                );
                assert!(
                    error.to_string().contains("maxLifetimeSeconds"),
                    "{platform} {declared} limits"
                );
            }

            // Where the field is unsupported, that is the answer, not the value.
            for platform in [Platform::Azure, Platform::Local] {
                let error = sandbox
                    .validate_for_platform(platform)
                    .expect_err("a lifetime is refused where nothing applies it");
                assert_eq!(
                    error.code, "SANDBOX_CAPABILITY_UNSUPPORTED",
                    "{platform} {declared} limits"
                );
            }

            sandbox.lifecycle.max_lifetime_seconds = Some(1);
            sandbox
                .validate_for_platform(Platform::Kubernetes)
                .expect("one second is the API server's floor");
        }
    }

    /// The API server caps `activeDeadlineSeconds` at the largest i32 although the field is an
    /// int64, and `maxLifetimeSeconds` is a u32, so without this check the top half of its
    /// range would plan fine and be refused when the pod is created.
    #[test]
    fn a_lifetime_kubernetes_would_reject_is_refused_while_planning() {
        let with_limits = sandbox_with(SandboxEgress::None, vec![]);
        let mut without_limits = with_limits.clone();
        without_limits.limits = None;

        for (declared, mut sandbox) in [("with", with_limits), ("without", without_limits)] {
            for seconds in [2_147_483_648, u32::MAX] {
                sandbox.lifecycle.max_lifetime_seconds = Some(seconds);
                let error = sandbox
                    .validate_for_platform(Platform::Kubernetes)
                    .expect_err("a lifetime above what the API server takes is refused");
                assert_eq!(
                    error.code, "SANDBOX_LIMIT_INVALID",
                    "{seconds}s {declared} limits"
                );
                assert!(
                    error.to_string().contains("maxLifetimeSeconds"),
                    "{seconds}s {declared} limits"
                );
            }

            sandbox.lifecycle.max_lifetime_seconds = Some(2_147_483_647);
            sandbox
                .validate_for_platform(Platform::Kubernetes)
                .expect("the ceiling itself is allowed");
        }
    }

    /// An image Azure can take neither as a catalog name nor as a registry image is refused while
    /// planning, not at the first sandbox.
    #[test]
    fn an_image_azure_cannot_pull_is_refused_while_planning() {
        let mut sandbox = sandbox_with(SandboxEgress::None, vec![]);
        // Azure enforces no declared ceiling, so a sandbox carrying limits is refused before the
        // image is ever read.
        sandbox.limits = None;

        // A digest too short to be one, blanks, a space and a query string.
        for image in ["ubuntu@sha256:abc", "", "   ", "ubuntu latest", "ubuntu?x"] {
            sandbox.code = SandboxCode::Image {
                image: image.to_string(),
                bundled_from: None,
            };
            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}'");
        }

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

        for image in [
            "ubuntu:24.04",
            "ghcr.io/myorg/sandbox:latest",
            "docker.io/library/python:3.14-slim",
            "localhost:5000/team/agent@sha256:51dafde81dbdb6ebde285137a295cf18a47ca95234fe388a343719cb97305b3d",
        ] {
            sandbox.code = SandboxCode::Image {
                image: image.to_string(),
                bundled_from: None,
            };
            sandbox
                .validate_for_platform(Platform::Azure)
                .unwrap_or_else(|error| panic!("'{image}' is a registry image: {error}"));
        }

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

    /// Every value the catalog allowlist `[A-Za-z0-9._-]+` accepts stays a catalog name, so a
    /// declaration that planned under it keeps its meaning. Exhaustive to three characters, then a
    /// fixed-seed sample of longer ones, some with surrounding space.
    #[test]
    fn every_value_the_catalog_allowlist_accepted_is_still_a_catalog_name() {
        const ALPHABET: &[u8] =
            b"abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789._-";

        let assert_catalog = |value: &str| {
            assert_eq!(
                classify_azure_sandbox_image(value),
                Some(AzureSandboxImage::Catalog(value.trim())),
                "'{value}'"
            );
        };

        for a in ALPHABET {
            assert_catalog(&String::from_utf8(vec![*a]).unwrap());
            for b in ALPHABET {
                assert_catalog(&String::from_utf8(vec![*a, *b]).unwrap());
                for c in ALPHABET {
                    assert_catalog(&String::from_utf8(vec![*a, *b, *c]).unwrap());
                }
            }
        }

        let mut state: u64 = 0x9E37_79B9_7F4A_7C15;
        let mut next = || {
            state ^= state << 13;
            state ^= state >> 7;
            state ^= state << 17;
            state
        };
        for _ in 0..20_000 {
            let len = 4 + (next() % 60) as usize;
            let mut value: String = (0..len)
                .map(|_| ALPHABET[(next() % ALPHABET.len() as u64) as usize] as char)
                .collect();
            if next() % 4 == 0 {
                value = format!("  {value}\t");
            }
            assert_catalog(&value);
        }
    }

    /// The label is how the provider finds the image the controller built, so both must derive
    /// the same one; it also has to fit a label value, which a raw reference does not.
    #[test]
    fn the_disk_image_label_is_stable_and_label_safe() {
        let label = azure_disk_image_label("docker.io/library/python:3.14-slim");
        assert_eq!(
            label,
            azure_disk_image_label(" docker.io/library/python:3.14-slim ")
        );
        assert_ne!(
            label,
            azure_disk_image_label("docker.io/library/python:3.13-slim")
        );
        assert_eq!(label.len(), 32);
        assert!(label.chars().all(|c| c.is_ascii_hexdigit()), "{label}");
    }

    /// 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::None, 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::None, 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::None, 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::None, 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}"
        );
    }

    /// Source is refused at plan time on every platform whose sandbox image `alien build` does
    /// not build (the reason is at the check in `validate_for_platform`).
    #[test]
    fn source_code_is_refused_where_no_backend_builds_it() {
        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::None)
            .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`");
        sandbox
            .validate_for_platform(Platform::Azure)
            .expect("an Azure sandbox base image is built by `alien build`");
        sandbox
            .validate_for_platform(Platform::Gcp)
            .expect("a GCP sandbox base image is built by `alien build`");

        for platform in [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 privileged_supervisor_fixes_identity_and_requires_an_enforceable_backend() {
        let mut sandbox = sandbox_with(
            SandboxEgress::AllowDomains {
                domains: vec!["example.com".to_string()],
            },
            vec![],
        );
        sandbox.privileged_supervisor = Some(SandboxPrivilegedSupervisor { command_uid: 60001 });
        sandbox
            .validate_for_platform(Platform::Aws)
            .expect("AWS can enforce the policy in the agent");
        assert_eq!(sandbox.cloud_egress(), &SandboxEgress::Allow);
        assert_eq!(
            sandbox.supervisor_environment()["ALIEN_SANDBOX_EXPECTED_EXEC_UID"],
            "60001"
        );
        for platform in [
            Platform::Local,
            Platform::Kubernetes,
            Platform::Azure,
            Platform::Gcp,
        ] {
            let error = sandbox
                .validate_for_platform(platform)
                .expect_err("cannot grant a privilege boundary the backend lacks");
            assert_eq!(error.code, "SANDBOX_CAPABILITY_UNSUPPORTED");
            assert!(error.message.contains("privilegedSupervisor"));
        }
        for uid in [0, u32::MAX] {
            sandbox.privileged_supervisor.as_mut().unwrap().command_uid = uid;
            let error = sandbox
                .validate_for_platform(Platform::Aws)
                .expect_err("invalid uid must fail at plan time");
            assert_eq!(error.code, "SANDBOX_LIMIT_INVALID");
        }
    }

    #[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::None, vec![]);
        let renamed = Sandbox::new("other".to_string())
            .code(SandboxCode::Image {
                image: "ubuntu".to_string(),
                bundled_from: None,
            })
            .limits(
                original
                    .limits
                    .clone()
                    .expect("the fixture declares limits"),
            )
            .egress(SandboxEgress::None)
            .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(),
                    bundled_from: None,
                })
                .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(),
                    bundled_from: None,
                })
                .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::None.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"
        );
    }

    /// A sandbox resource as a stack state recorded it before `session` became `lifecycle` and
    /// `idleSuspendSeconds` became `idlePauseSeconds`. Such a state is still what a deployment
    /// that old holds, and reading it is the first step of updating or deleting that deployment.
    #[test]
    fn stack_state_written_before_the_lifecycle_rename_still_reads() {
        let stored: crate::StackResourceState = serde_json::from_value(serde_json::json!({
            "type": "sandbox",
            "config": {
                "id": "sandbox",
                "code": {
                    "type": "image",
                    "image": "s3://example-bundles/analysis/v1/bundle.zip"
                },
                "type": "sandbox",
                "egress": { "mode": "allow" },
                "session": { "maxLifetimeSeconds": 28800, "idleSuspendSeconds": 600 }
            },
            "status": "running",
            "outputs": {
                "type": "sandbox",
                "identifier": "arn:aws:lambda:us-east-2:123456789012:microvm-image:example-sandbox",
                "parentName": "arn:aws:lambda:us-east-2:123456789012:microvm-image:example-sandbox"
            },
            "lifecycle": "frozen",
            "dependencies": [
                { "id": "management", "type": "remote-stack-management" },
                { "id": "access", "type": "resource-access" }
            ],
            "controllerPlatform": "aws"
        }))
        .expect("a stack state written before the rename must still deserialize");

        let expected = Sandbox::new("sandbox".to_string())
            .code(SandboxCode::Image {
                image: "s3://example-bundles/analysis/v1/bundle.zip".to_string(),
                bundled_from: None,
            })
            .egress(SandboxEgress::Allow)
            .lifecycle(SandboxLifecyclePolicy {
                max_lifetime_seconds: Some(28_800),
                idle_pause_seconds: Some(600),
            })
            .build();
        assert_eq!(stored.config.downcast_ref::<Sandbox>(), Some(&expected));

        let rewritten = serde_json::to_value(&stored.config).expect("the sandbox serializes");
        assert_eq!(
            rewritten["lifecycle"],
            serde_json::json!({ "maxLifetimeSeconds": 28800, "idlePauseSeconds": 600 }),
            "the next write stores the current names"
        );
        assert!(rewritten.get("session").is_none());
    }
}