alien_core/

compute_planner.rs

//! Deployment-time compute planner.
//!
//! The planner turns portable stack requirements plus a target platform into a
//! renderable set of recommended deployment choices. It does not mutate the
//! stack and does not require database access.

use crate::{
    instance_catalog::{self, Architecture, WorkloadRequirements},
    CapacityGroup, CapacityGroupScalePolicy, ComputeChoiceRange, ComputePoolSelection, Container,
    Daemon, ErrorData, FailureDomainSelection, GpuSpec, MachineProfile, Platform, ResourceSpec,
    Stack, StackState,
};
use alien_error::{AlienError, Result};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;

/// Full compute plan for one stack/platform pair.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(rename_all = "camelCase")]
pub struct ComputePlan {
    /// Planned pools in stable pool-id order.
    pub pools: Vec<ComputePoolPlan>,
    /// Containers whose per-replica allocation can be chosen during setup.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub containers: Vec<ContainerResourcePlan>,
}

/// Deployment-time resource controls and effective allocation for one container.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(rename_all = "camelCase")]
pub struct ContainerResourcePlan {
    /// Stable container resource ID.
    pub container_id: String,
    /// Allowed allocations declared in the release.
    pub choices: crate::ContainerResourceChoices,
    /// Effective per-replica CPU reservation and limit.
    pub cpu: ResourceSpec,
    /// Effective per-replica memory reservation and limit.
    pub memory: ResourceSpec,
    /// Effective persistent capacity per replica, in GiB.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub volume_size_gib: Option<u32>,
    /// Installed capacity, or the provider minimum for a new volume.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub minimum_volume_size_gib: Option<u32>,
    /// Maximum supported capacity for this provider, in GiB.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub maximum_volume_size_gib: Option<u32>,
}

/// Planner output for one compute pool.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(rename_all = "camelCase")]
pub struct ComputePoolPlan {
    /// Pool ID from the stack or derived default.
    pub pool_id: String,
    /// Workloads assigned to this pool.
    pub workloads: Vec<String>,
    /// Aggregated requirements used for machine selection.
    pub requirements: MachineProfile,
    /// Allowed scale policy declared by source or derived for generated pools.
    pub scale: CapacityGroupScalePolicy,
    /// Recommended or user-selected deployment choice.
    pub selected: ComputePoolSelection,
    /// Planner-recommended default.
    pub recommended: ComputePoolSelection,
    /// Valid cloud machine choices. Empty for local and Kubernetes.
    pub machines: Vec<ComputeMachineOption>,
    /// Data disk sizes this deployment may choose, when the pool declares a data disk.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub data_disk: Option<ComputePoolDataDiskPlan>,
    /// Validation errors for supplied deployment settings.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub errors: Vec<String>,
}

/// Data disk sizes a deployment may choose for one pool, in whole GiB. A fixed
/// declaration has `min_gib == max_gib`. A size of 0 runs the pool without a
/// data disk, which only a range starting at 0 allows.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(rename_all = "camelCase")]
pub struct ComputePoolDataDiskPlan {
    /// Smallest size the deployment may choose.
    pub min_gib: u32,
    /// Largest size the deployment may choose.
    pub max_gib: u32,
    /// Size used when the deployment does not choose one.
    pub default_gib: u32,
    /// Size this deployment runs with: its choice, or the default.
    pub selected_gib: u32,
}

/// One concrete provider machine option.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
#[serde(rename_all = "camelCase")]
pub struct ComputeMachineOption {
    /// Provider machine name.
    pub machine: String,
    /// Machine hardware profile.
    pub profile: MachineProfile,
    /// Whether this machine is the planner's default recommendation.
    pub recommended: bool,
}

/// Failure-domain policy for a pool that hosts persistent stateful workloads when the
/// deployment does not choose one: one provider-selected domain, so the pool's machines
/// and their volumes land in the same zone.
pub fn default_persistent_failure_domains() -> FailureDomainSelection {
    FailureDomainSelection {
        spread: 1,
        selected_failure_domains: Vec::new(),
    }
}

/// Compute a deterministic deployment-time plan.
pub fn plan_compute(
    stack: &Stack,
    platform: Platform,
    selected_settings: Option<&crate::ComputeSettings>,
) -> Result<ComputePlan, ErrorData> {
    plan_compute_with_state(stack, platform, selected_settings, None)
}

/// Plan a configuration update while retaining the installed machine architecture.
/// Resource changes must not silently require a different workload image target.
pub fn plan_compute_with_state(
    stack: &Stack,
    platform: Platform,
    selected_settings: Option<&crate::ComputeSettings>,
    previous_state: Option<&StackState>,
) -> Result<ComputePlan, ErrorData> {
    let mut resolved_stack = stack.clone();
    crate::container_resources::resolve_container_resources(
        &mut resolved_stack,
        selected_settings,
    )?;
    if let Some(previous_state) = previous_state {
        crate::container_resources::retain_installed_volume_sizes(
            &mut resolved_stack,
            selected_settings,
            previous_state,
        );
    }
    let (minimum, maximum) = match platform {
        Platform::Gcp => (10, 65536),
        Platform::Azure => (1, 32767),
        _ => (1, 65536),
    };
    for (id, entry) in &resolved_stack.resources {
        let Some(storage) = entry
            .config
            .downcast_ref::<Container>()
            .and_then(|container| container.persistent_storage.as_ref())
        else {
            continue;
        };
        let size = crate::container_resources::parse_volume_size_gib(&storage.size)
            .map_err(|reason| crate::container_resources::invalid(id, "volumeSizeGib", &reason))?;
        if size < minimum || size > maximum {
            return Err(AlienError::new(
                ErrorData::ContainerResourceSelectionInvalid {
                    resource_id: id.clone(),
                    dimension: "volumeSizeGib".into(),
                    reason: format!(
                        "capacity for {platform} must be between {minimum} and {maximum} GiB"
                    ),
                },
            ));
        }
    }
    if let Some(previous_state) = previous_state {
        for (id, entry) in &resolved_stack.resources {
            let desired = entry
                .config
                .downcast_ref::<Container>()
                .and_then(|container| container.persistent_storage.as_ref());
            let previous = previous_state
                .resources
                .get(id)
                .and_then(|resource| resource.config.downcast_ref::<Container>())
                .and_then(|container| container.persistent_storage.as_ref());
            if previous.is_some() && desired.is_none() {
                return Err(crate::container_resources::invalid(
                    id,
                    "volumeSizeGib",
                    "removing installed persistent storage is unsupported",
                ));
            }
            if let (Some(desired), Some(previous)) = (desired, previous) {
                if desired.mount_path != previous.mount_path {
                    return Err(crate::container_resources::invalid(
                        id,
                        "volumeSizeGib",
                        "changing an installed persistent storage mount path is unsupported",
                    ));
                }
                let target = crate::container_resources::parse_volume_size_gib(&desired.size);
                let installed = crate::container_resources::parse_volume_size_gib(&previous.size);
                if let (Ok(target), Ok(installed)) = (target, installed) {
                    if target < installed {
                        return Err(AlienError::new(ErrorData::ContainerResourceSelectionInvalid {
                            resource_id: id.clone(), dimension: "volumeSizeGib".into(),
                            reason: format!("existing volumes cannot shrink from {installed} to {target} GiB"),
                        }));
                    }
                }
            }
        }
    }
    let stack = &resolved_stack;
    let mut groups = collect_workload_groups(stack)?;
    merge_explicit_compute_groups(stack, &mut groups)?;

    let mut pool_ids: Vec<String> = groups.keys().cloned().collect();
    pool_ids.sort();

    let mut pools = Vec::new();
    for pool_id in pool_ids {
        let mut group = groups.remove(&pool_id).expect("pool id came from map keys");
        let mut requirements = group.requirements;
        let selected = selected_settings.and_then(|settings| settings.pools.get(&pool_id));
        let declared_group = stack
            .resources
            .values()
            .filter_map(|entry| entry.config.downcast_ref::<crate::ComputeCluster>())
            .flat_map(|cluster| &cluster.capacity_groups)
            .find(|group| group.group_id == pool_id);
        let declared_machine = declared_group.and_then(|group| group.instance_type.as_deref());
        let machine_override = match selected {
            Some(selection) => selection.machine(),
            None => declared_machine,
        };
        if requirements.architecture.is_none() {
            requirements.architecture = machine_override
                .and_then(|machine| instance_catalog::find_instance_type(platform, machine))
                .map(|spec| spec.architecture)
                .or_else(|| {
                    previous_state.and_then(|state| {
                        state
                            .resources
                            .values()
                            .filter_map(|resource| {
                                resource.config.downcast_ref::<crate::ComputeCluster>()
                            })
                            .flat_map(|cluster| &cluster.capacity_groups)
                            .find(|group| group.group_id == pool_id)
                            .and_then(|group| group.instance_type.as_deref())
                            .and_then(|machine| {
                                instance_catalog::find_instance_type(platform, machine)
                            })
                            .map(|spec| spec.architecture)
                    })
                });
        }
        if group.generated {
            group.scale = generated_pool_scale_policy(
                group.scale.default_min_size(),
                group.scale.default_max_size(),
                selected.map(ComputePoolSelection::max_size),
            );
        }
        let machines = selected
            .map(ComputePoolSelection::max_size)
            .unwrap_or(group.scale.default_max_size())
            .max(1);
        let mut sizing = requirements.clone();
        // A bounded fleet must fit total demand as well as its largest replica.
        sizing.max_cpu_per_container = sizing
            .max_cpu_per_container
            .max(sizing.total_cpu_at_desired / f64::from(machines));
        sizing.max_memory_per_container = sizing.max_memory_per_container.max(
            sizing
                .total_memory_bytes_at_desired
                .div_ceil(u64::from(machines)),
        );
        let recommended = recommended_selection(
            platform,
            &sizing,
            &group.scale,
            group.requires_failure_domain,
            group.generated,
            machine_override,
        )?;
        let mut selected_choice = selected.cloned().unwrap_or_else(|| recommended.clone());
        if selected.is_none() {
            if let Some(declared_machine) = declared_machine {
                match &mut selected_choice {
                    ComputePoolSelection::Fixed { machine, .. }
                    | ComputePoolSelection::Autoscale { machine, .. } => {
                        *machine = Some(declared_machine.to_string());
                    }
                }
            }
        }
        // An omitted machine is deployment intent: let the planner choose on
        // every preparation. Only the returned plan contains the concrete type;
        // never write it back into the caller's settings as an override.
        match &mut selected_choice {
            ComputePoolSelection::Fixed { machine, .. }
            | ComputePoolSelection::Autoscale { machine, .. } => {
                if machine.is_none() {
                    *machine = recommended.machine().map(ToString::to_string);
                }
            }
        }
        if selected_choice.failure_domains().is_none() {
            if let Some(default_failure_domains) = recommended.failure_domains().cloned() {
                match &mut selected_choice {
                    ComputePoolSelection::Fixed {
                        failure_domains, ..
                    }
                    | ComputePoolSelection::Autoscale {
                        failure_domains, ..
                    } => *failure_domains = Some(default_failure_domains),
                }
            }
        }
        let mut errors = validate_compute_pool_selection(
            platform,
            &pool_id,
            &selected_choice,
            &requirements,
            &group.scale,
        );
        // Only workloads assigned to the pool give it real demand; a declared pool without
        // them carries just its machine profile.
        if let Some(workload_requirements) = &group.workload_requirements {
            if let Err(message) =
                check_pool_capacity(platform, &pool_id, &selected_choice, workload_requirements)
            {
                errors.push(message);
            }
        }
        let selected_data_disk_gib = selected.and_then(ComputePoolSelection::data_disk_gib);
        let data_disk_choices = declared_group.and_then(crate::CapacityGroup::data_disk_choices);
        let data_disk_result = match declared_group {
            Some(group) => group.effective_data_disk_gib(selected_data_disk_gib),
            None => match selected_data_disk_gib {
                None | Some(0) => Ok(None),
                Some(size) => Err(format!(
                    "pool '{pool_id}' declares no data disk, so dataDiskGib {size} cannot be chosen"
                )),
            },
        };
        if let Err(message) = &data_disk_result {
            errors.push(message.clone());
        }
        let data_disk = data_disk_choices.map(|choices| ComputePoolDataDiskPlan {
            min_gib: choices.min,
            max_gib: choices.max,
            default_gib: choices.default,
            selected_gib: match &data_disk_result {
                Ok(size) => size.unwrap_or(0),
                Err(_) => selected_data_disk_gib.unwrap_or(choices.default),
            },
        });
        let machines = machine_options(
            platform,
            &requirements,
            selected_choice.machine(),
            recommended.machine(),
        )?;

        pools.push(ComputePoolPlan {
            pool_id,
            workloads: group.workloads,
            requirements: requirements_to_profile(&requirements),
            scale: group.scale,
            selected: selected_choice,
            recommended,
            machines,
            data_disk,
            errors,
        });
    }

    let mut containers: Vec<_> = stack
        .resources
        .values()
        .filter_map(|entry| entry.config.downcast_ref::<Container>())
        .filter(|container| {
            container.resource_choices.is_some() || container.persistent_storage.is_some()
        })
        .map(|container| {
            let capacity = |storage: &crate::PersistentStorage| {
                crate::container_resources::parse_volume_size_gib(&storage.size).map_err(|reason| {
                    crate::container_resources::invalid(&container.id, "volumeSizeGib", &reason)
                })
            };
            let volume_size_gib = container
                .persistent_storage
                .as_ref()
                .map(capacity)
                .transpose()?;
            let installed = previous_state
                .and_then(|state| state.resources.get(&container.id))
                .and_then(|resource| resource.config.downcast_ref::<Container>())
                .and_then(|container| container.persistent_storage.as_ref())
                .map(capacity)
                .transpose()?;
            Ok(ContainerResourcePlan {
                container_id: container.id.clone(),
                choices: container.resource_choices.clone().unwrap_or_default(),
                cpu: container.cpu.clone(),
                memory: container.memory.clone(),
                volume_size_gib,
                minimum_volume_size_gib: volume_size_gib
                    .map(|_| installed.unwrap_or(if platform == Platform::Gcp { 10 } else { 1 })),
                maximum_volume_size_gib: volume_size_gib.map(|_| {
                    if platform == Platform::Azure {
                        32767
                    } else {
                        65536
                    }
                }),
            })
        })
        .collect::<Result<Vec<_>, ErrorData>>()?;
    containers.sort_by(|a, b| a.container_id.cmp(&b.container_id));
    Ok(ComputePlan { pools, containers })
}

#[derive(Debug, Clone)]
struct PlannedGroup {
    workloads: Vec<String>,
    /// Actual allocations, excluding a declaration's minimum machine profile.
    workload_requirements: Option<WorkloadRequirements>,
    requirements: WorkloadRequirements,
    scale: CapacityGroupScalePolicy,
    requires_failure_domain: bool,
    /// No capacity group in the source declares this pool. Its scale is the
    /// planner's recommendation, not a bound the developer chose.
    generated: bool,
}

/// Machine count up to which a setup offers choices for a pool the source does not declare.
///
/// It is not a limit: such a pool has no upper bound, and a saved choice above it widens the
/// offered range. Recommendations above it are clamped to it.
pub const GENERATED_POOL_MAX_MACHINES: u32 = 100;

/// Allowed scale for a pool that no capacity group declares.
///
/// The planner derives `default_min`/`default_max` from workload replicas, assuming one
/// container per machine. They are a recommendation, not a bound: the developer did not bound
/// the pool, a larger machine fits more containers, and replica counts change between releases
/// while the installer's saved choice stays. The installer may choose any count from the
/// workload minimum (one machine when any workload runs) upwards, fixed or autoscaling (a fixed
/// N is accepted as the autoscale range N..N). The range reaches
/// [`GENERATED_POOL_MAX_MACHINES`], or `selected_max` when the saved choice is larger, so the
/// saved choice is always inside it. Whether the choice can hold the workloads is
/// [`check_pool_capacity`]'s question.
pub fn generated_pool_scale_policy(
    default_min: u32,
    default_max: u32,
    selected_max: Option<u32>,
) -> CapacityGroupScalePolicy {
    let offered = GENERATED_POOL_MAX_MACHINES;
    let ceiling = offered.max(selected_max.unwrap_or(0));
    let default_min = default_min.min(offered);
    CapacityGroupScalePolicy::Autoscale {
        min: ComputeChoiceRange {
            min: default_min,
            max: ceiling,
            default: default_min,
        },
        max: ComputeChoiceRange {
            min: default_min.max(1),
            max: ceiling,
            default: default_max.clamp(default_min.max(1), offered),
        },
    }
}

/// Rejects a selection whose largest fleet cannot hold the workloads at their desired replica
/// counts after reserving host resources. It also checks that one replica fits;
/// arbitrary multi-replica packing can still require runtime validation.
pub fn check_pool_capacity(
    platform: Platform,
    pool_id: &str,
    selection: &ComputePoolSelection,
    requirements: &WorkloadRequirements,
) -> std::result::Result<(), String> {
    let Some(spec) = selection
        .machine()
        .and_then(|machine| instance_catalog::find_instance_type(platform, machine))
    else {
        // Machine presence and catalog membership are reported by the selection checks.
        return Ok(());
    };
    let machines = selection.max_size();
    let per_machine_cpu = instance_catalog::allocatable_cpu(spec);
    let per_machine_memory = instance_catalog::allocatable_memory_bytes(spec);
    if per_machine_cpu + f64::EPSILON < requirements.max_cpu_per_container
        || per_machine_memory < requirements.max_memory_per_container
    {
        return Err(format!(
            "Pool '{pool_id}' machine '{}' cannot fit one replica after reserving host resources",
            spec.name,
        ));
    }
    let cpu = f64::from(machines) * per_machine_cpu;
    let memory = u64::from(machines) * per_machine_memory;
    if cpu + f64::EPSILON >= requirements.total_cpu_at_desired
        && memory >= requirements.total_memory_bytes_at_desired
    {
        return Ok(());
    }
    const GIB: f64 = 1024.0 * 1024.0 * 1024.0;
    Err(format!(
        "Pool '{pool_id}' is too small for its workloads: {machines} x {} has {cpu:.2} vCPU and {:.1} GiB, and the workloads request {:.2} vCPU and {:.1} GiB at their desired replica counts",
        spec.name,
        memory as f64 / GIB,
        requirements.total_cpu_at_desired,
        requirements.total_memory_bytes_at_desired as f64 / GIB,
    ))
}

fn collect_workload_groups(stack: &Stack) -> Result<HashMap<String, PlannedGroup>, ErrorData> {
    let mut groups: HashMap<String, Vec<Workload>> = HashMap::new();

    for entry in stack.resources.values() {
        if let Some(container) = entry.config.downcast_ref::<Container>() {
            groups
                .entry(
                    container
                        .pool
                        .clone()
                        .unwrap_or_else(|| needed_container_pool(container).to_string()),
                )
                .or_default()
                .push(Workload::from_container(container)?);
        }
        if let Some(daemon) = entry.config.downcast_ref::<Daemon>() {
            if daemon.cluster.is_some() {
                groups
                    .entry(daemon.pool.clone().unwrap_or_else(|| "general".to_string()))
                    .or_default()
                    .push(Workload::from_daemon(daemon)?);
            }
        }
    }

    let mut planned = HashMap::new();
    for (pool_id, workloads) in groups {
        let requirements = aggregate_workloads(&workloads);
        let requires_failure_domain = workloads
            .iter()
            .any(|workload| workload.requires_failure_domain);
        let min_size = default_min_machines(&requirements);
        let max_size = default_max_machines(&requirements);
        planned.insert(
            pool_id,
            PlannedGroup {
                workloads: workloads.into_iter().map(|w| w.id).collect(),
                workload_requirements: Some(requirements.clone()),
                // Merged with any declared group below; widened in `plan_compute` if none declares it.
                scale: CapacityGroupScalePolicy::from_selected_bounds(min_size, max_size),
                requirements,
                requires_failure_domain,
                generated: true,
            },
        );
    }
    Ok(planned)
}

fn merge_explicit_compute_groups(
    stack: &Stack,
    groups: &mut HashMap<String, PlannedGroup>,
) -> Result<(), ErrorData> {
    for entry in stack.resources.values() {
        let Some(cluster) = entry.config.downcast_ref::<crate::ComputeCluster>() else {
            continue;
        };
        for group in &cluster.capacity_groups {
            let explicit_requirements = profile_to_requirements(
                group.profile.as_ref(),
                group.nested_virtualization.unwrap_or(false),
            );
            let scale = group.scale_policy.clone().unwrap_or_else(|| {
                CapacityGroupScalePolicy::from_selected_bounds(group.min_size, group.max_size)
            });
            groups
                .entry(group.group_id.clone())
                .and_modify(|planned| {
                    if group.instance_type.is_none() {
                        merge_requirements(&mut planned.requirements, &explicit_requirements);
                    } else {
                        // A materialized capacity group profile describes the
                        // selected machine, not additional workload demand.
                        // Merging its full CPU and memory into requirements
                        // would cause deployment package planning to size an
                        // already-sized machine a second time. Keep only the
                        // constraints that affect artifact compatibility.
                        planned.requirements.architecture = planned
                            .requirements
                            .architecture
                            .or(explicit_requirements.architecture);
                        planned.requirements.nested_virt |= explicit_requirements.nested_virt;
                    }
                    planned.scale = merge_scale_policy(&planned.scale, &scale);
                    planned.generated = false;
                })
                .or_insert_with(|| PlannedGroup {
                    workloads: Vec::new(),
                    workload_requirements: None,
                    scale,
                    requirements: explicit_requirements,
                    requires_failure_domain: false,
                    generated: false,
                });
        }
    }
    Ok(())
}

fn recommended_selection(
    platform: Platform,
    requirements: &WorkloadRequirements,
    scale: &CapacityGroupScalePolicy,
    requires_failure_domain: bool,
    generated: bool,
    machine_override: Option<&str>,
) -> Result<ComputePoolSelection, ErrorData> {
    let machine = match platform {
        Platform::Aws | Platform::Gcp | Platform::Azure => Some(
            match instance_catalog::select_instance_type(platform, requirements) {
                Ok(selection) => selection.instance_type.to_string(),
                Err(message) => {
                    // Automatic recommendations add headroom. A valid explicit
                    // machine remains usable even when no catalog entry can
                    // provide that extra headroom.
                    let fallback = machine_override
                        .and_then(|machine| instance_catalog::find_instance_type(platform, machine))
                        .filter(|spec| {
                            instance_satisfies(
                                spec,
                                requirements,
                                requirements.architecture.unwrap_or(spec.architecture),
                            )
                        });
                    fallback.map(|spec| spec.name.to_string()).ok_or_else(|| {
                        AlienError::new(ErrorData::GenericError {
                            message: format!("Failed to select {platform} machine: {message}"),
                        })
                    })?
                }
            },
        ),
        Platform::Local | Platform::Kubernetes | Platform::Machines | Platform::Test => None,
    };

    let failure_domains = (requires_failure_domain
        && matches!(platform, Platform::Aws | Platform::Gcp | Platform::Azure))
    .then(default_persistent_failure_domains);

    match scale {
        CapacityGroupScalePolicy::Fixed { machines } => Ok(ComputePoolSelection::Fixed {
            machines: machines.default.max(1),
            machine,
            failure_domains,
            data_disk_gib: None,
        }),
        // Workloads that never scale need no autoscaling fleet, even where any mode is allowed.
        CapacityGroupScalePolicy::Autoscale { min, max }
            if generated && min.default == max.default =>
        {
            Ok(ComputePoolSelection::Fixed {
                machines: min.default.max(1),
                machine,
                failure_domains,
                data_disk_gib: None,
            })
        }
        CapacityGroupScalePolicy::Autoscale { min, max } => Ok(ComputePoolSelection::Autoscale {
            min: min.default,
            max: max.default.max(min.default),
            machine,
            failure_domains,
            data_disk_gib: None,
        }),
    }
}

/// Validate one selected compute pool against platform machine requirements and
/// source-declared scale bounds.
pub fn validate_compute_pool_selection(
    platform: Platform,
    pool_id: &str,
    selection: &ComputePoolSelection,
    requirements: &WorkloadRequirements,
    scale: &CapacityGroupScalePolicy,
) -> Vec<String> {
    let mut errors = Vec::new();
    if let Err(message) = selection.validate() {
        errors.push(message);
    }
    if let Err(message) = validate_selection_against_scale(selection, scale) {
        errors.push(format!("Pool '{pool_id}' {message}"));
    }
    if matches!(platform, Platform::Aws | Platform::Gcp | Platform::Azure) {
        match selection.machine() {
            Some(machine) => match instance_catalog::find_instance_type(platform, machine) {
                Some(spec) => {
                    let architecture = requirements.architecture.unwrap_or(spec.architecture);
                    if !instance_satisfies(spec, requirements, architecture) {
                        errors.push(format!(
                            "{} machine '{}' does not satisfy pool '{}' requirements",
                            platform, machine, pool_id
                        ));
                    }
                }
                None => errors.push(format!(
                    "Unknown {} machine '{}' for pool '{}'",
                    platform, machine, pool_id
                )),
            },
            None => errors.push(format!(
                "Pool '{}' requires a provider machine on {}",
                pool_id, platform
            )),
        }
    }
    errors
}

/// Convert a capacity group declaration into planner requirements.
pub fn capacity_group_requirements(group: &CapacityGroup) -> WorkloadRequirements {
    profile_to_requirements(
        group.profile.as_ref(),
        group.nested_virtualization.unwrap_or(false),
    )
}

fn machine_options(
    platform: Platform,
    requirements: &WorkloadRequirements,
    selected_machine: Option<&str>,
    recommended_machine: Option<&str>,
) -> Result<Vec<ComputeMachineOption>, ErrorData> {
    if !matches!(platform, Platform::Aws | Platform::Gcp | Platform::Azure) {
        return Ok(Vec::new());
    }
    let resolved_architecture = requirements
        .architecture
        .or_else(|| {
            selected_machine.and_then(|machine| {
                instance_catalog::find_instance_type(platform, machine)
                    .map(|spec| spec.architecture)
            })
        })
        .or_else(|| {
            recommended_machine
                .and_then(|machine| instance_catalog::find_instance_type(platform, machine))
                .map(|spec| spec.architecture)
        })
        .ok_or_else(|| {
            AlienError::new(ErrorData::GenericError {
                message: format!("Selected {platform} machine has no CPU architecture"),
            })
        })?;

    let mut options: Vec<ComputeMachineOption> = instance_catalog::catalog_for_platform(platform)
        .into_iter()
        .filter(|spec| instance_satisfies(spec, requirements, resolved_architecture))
        .map(|spec| ComputeMachineOption {
            machine: spec.name.to_string(),
            profile: spec.to_machine_profile_for_storage(requirements.max_ephemeral_storage_bytes),
            recommended: Some(spec.name) == recommended_machine
                || Some(spec.name) == selected_machine,
        })
        .collect();
    options.sort_by(|a, b| a.machine.cmp(&b.machine));
    Ok(options)
}

fn instance_satisfies(
    spec: &instance_catalog::InstanceTypeSpec,
    requirements: &WorkloadRequirements,
    resolved_architecture: Architecture,
) -> bool {
    if spec.architecture != resolved_architecture {
        return false;
    }
    if requirements.nested_virt && !spec.is_nested_virt_capable() {
        return false;
    }
    if spec.vcpu < requirements.max_cpu_per_container.ceil() as u32 {
        return false;
    }
    if spec.memory_bytes < requirements.max_memory_per_container {
        return false;
    }
    if spec.has_configurable_ephemeral_storage() {
        if crate::instance_catalog::max_configurable_ephemeral_storage_bytes(spec.platform)
            .is_none_or(|max| requirements.max_ephemeral_storage_bytes > max)
        {
            return false;
        }
    } else if spec.ephemeral_storage_bytes < requirements.max_ephemeral_storage_bytes {
        return false;
    }
    match (&requirements.gpu, spec.gpu) {
        (Some(required), Some(actual)) => {
            (required.gpu_type == "any" || required.gpu_type == actual.gpu_type)
                && actual.count >= required.count
        }
        (Some(_), None) => false,
        (None, _) => true,
    }
}

#[derive(Debug, Clone)]
struct Workload {
    id: String,
    cpu: f64,
    memory_bytes: u64,
    desired_replicas: f64,
    max_replicas: f64,
    ephemeral_storage_bytes: u64,
    gpu: Option<GpuSpec>,
    requires_failure_domain: bool,
}

impl Workload {
    fn from_container(container: &Container) -> Result<Self, ErrorData> {
        let cpu = parse_cpu(&container.id, &container.cpu)?;
        let memory_bytes = parse_memory(&container.id, &container.memory)?;
        let desired_replicas = container
            .autoscaling
            .as_ref()
            .map(|a| a.desired)
            .or(container.replicas)
            .unwrap_or(1) as f64;
        let max_replicas = container
            .autoscaling
            .as_ref()
            .map(|a| a.max)
            .or(container.replicas)
            .unwrap_or(1) as f64;
        let ephemeral_storage_bytes = container
            .ephemeral_storage
            .as_deref()
            .map(instance_catalog::parse_memory_bytes)
            .transpose()
            .map_err(|message| {
                AlienError::new(ErrorData::GenericError {
                    message: format!(
                        "Failed to parse ephemeral storage for '{}': {message}",
                        container.id
                    ),
                })
            })?
            .unwrap_or(0);

        Ok(Self {
            id: container.id.clone(),
            cpu,
            memory_bytes,
            desired_replicas,
            max_replicas,
            ephemeral_storage_bytes,
            gpu: container.gpu.as_ref().map(|gpu| GpuSpec {
                gpu_type: gpu.gpu_type.clone(),
                count: gpu.count,
            }),
            requires_failure_domain: container.stateful && container.persistent_storage.is_some(),
        })
    }

    fn from_daemon(daemon: &Daemon) -> Result<Self, ErrorData> {
        Ok(Self {
            id: daemon.id.clone(),
            cpu: parse_cpu(&daemon.id, &daemon.cpu)?,
            memory_bytes: parse_memory(&daemon.id, &daemon.memory)?,
            desired_replicas: 1.0,
            max_replicas: 1.0,
            ephemeral_storage_bytes: 0,
            gpu: None,
            requires_failure_domain: false,
        })
    }
}

fn parse_cpu(resource_id: &str, spec: &ResourceSpec) -> Result<f64, ErrorData> {
    instance_catalog::parse_cpu(&spec.desired).map_err(|message| {
        AlienError::new(ErrorData::GenericError {
            message: format!(
                "Failed to parse CPU requirement '{}' for '{}': {message}",
                spec.desired, resource_id
            ),
        })
    })
}

fn parse_memory(resource_id: &str, spec: &ResourceSpec) -> Result<u64, ErrorData> {
    instance_catalog::parse_memory_bytes(&spec.desired).map_err(|message| {
        AlienError::new(ErrorData::GenericError {
            message: format!(
                "Failed to parse memory requirement '{}' for '{}': {message}",
                spec.desired, resource_id
            ),
        })
    })
}

fn aggregate_workloads(workloads: &[Workload]) -> WorkloadRequirements {
    let mut requirements = default_requirements();
    requirements.total_cpu_at_desired = 0.0;
    requirements.total_memory_bytes_at_desired = 0;
    requirements.total_cpu_at_max = 0.0;
    requirements.total_memory_bytes_at_max = 0;
    requirements.max_cpu_per_container = 0.0;
    requirements.max_memory_per_container = 0;
    requirements.max_ephemeral_storage_bytes = 0;
    requirements.gpu = None;

    for workload in workloads {
        requirements.total_cpu_at_desired += workload.cpu * workload.desired_replicas;
        requirements.total_cpu_at_max += workload.cpu * workload.max_replicas;
        requirements.total_memory_bytes_at_desired +=
            (workload.memory_bytes as f64 * workload.desired_replicas) as u64;
        requirements.total_memory_bytes_at_max +=
            (workload.memory_bytes as f64 * workload.max_replicas) as u64;
        requirements.max_cpu_per_container = requirements.max_cpu_per_container.max(workload.cpu);
        requirements.max_memory_per_container = requirements
            .max_memory_per_container
            .max(workload.memory_bytes);
        requirements.max_ephemeral_storage_bytes = requirements
            .max_ephemeral_storage_bytes
            .max(workload.ephemeral_storage_bytes);
        if requirements.gpu.is_none() {
            requirements.gpu = workload.gpu.clone();
        }
    }
    requirements
}

fn default_requirements() -> WorkloadRequirements {
    WorkloadRequirements {
        total_cpu_at_desired: 1.0,
        total_memory_bytes_at_desired: 2 * 1024 * 1024 * 1024,
        total_cpu_at_max: 1.0,
        total_memory_bytes_at_max: 2 * 1024 * 1024 * 1024,
        max_cpu_per_container: 1.0,
        max_memory_per_container: 2 * 1024 * 1024 * 1024,
        max_ephemeral_storage_bytes: 0,
        gpu: None,
        architecture: None,
        nested_virt: false,
    }
}

fn profile_to_requirements(
    profile: Option<&MachineProfile>,
    nested_virt: bool,
) -> WorkloadRequirements {
    let Some(profile) = profile else {
        return WorkloadRequirements {
            nested_virt,
            ..default_requirements()
        };
    };
    let cpu = instance_catalog::parse_cpu(&profile.cpu).unwrap_or(1.0);
    WorkloadRequirements {
        total_cpu_at_desired: cpu,
        total_memory_bytes_at_desired: profile.memory_bytes,
        total_cpu_at_max: cpu,
        total_memory_bytes_at_max: profile.memory_bytes,
        max_cpu_per_container: cpu,
        max_memory_per_container: profile.memory_bytes,
        max_ephemeral_storage_bytes: profile.ephemeral_storage_bytes,
        gpu: profile.gpu.clone(),
        architecture: profile.architecture,
        nested_virt,
    }
}

fn merge_requirements(existing: &mut WorkloadRequirements, declared: &WorkloadRequirements) {
    existing.total_cpu_at_desired = existing
        .total_cpu_at_desired
        .max(declared.total_cpu_at_desired);
    existing.total_memory_bytes_at_desired = existing
        .total_memory_bytes_at_desired
        .max(declared.total_memory_bytes_at_desired);
    existing.total_cpu_at_max = existing.total_cpu_at_max.max(declared.total_cpu_at_max);
    existing.total_memory_bytes_at_max = existing
        .total_memory_bytes_at_max
        .max(declared.total_memory_bytes_at_max);
    existing.max_cpu_per_container = existing
        .max_cpu_per_container
        .max(declared.max_cpu_per_container);
    existing.max_memory_per_container = existing
        .max_memory_per_container
        .max(declared.max_memory_per_container);
    existing.max_ephemeral_storage_bytes = existing
        .max_ephemeral_storage_bytes
        .max(declared.max_ephemeral_storage_bytes);
    if existing.gpu.is_none() {
        existing.gpu = declared.gpu.clone();
    }
    existing.nested_virt |= declared.nested_virt;
    if existing.architecture.is_none() {
        existing.architecture = declared.architecture;
    }
}

fn requirements_to_profile(requirements: &WorkloadRequirements) -> MachineProfile {
    MachineProfile {
        cpu: requirements.max_cpu_per_container.to_string(),
        memory_bytes: requirements.max_memory_per_container,
        ephemeral_storage_bytes: requirements.max_ephemeral_storage_bytes,
        architecture: requirements.architecture,
        gpu: requirements.gpu.clone(),
        data_disk: None,
    }
}

fn merge_scale_policy(
    existing: &CapacityGroupScalePolicy,
    declared: &CapacityGroupScalePolicy,
) -> CapacityGroupScalePolicy {
    match (existing, declared) {
        (
            CapacityGroupScalePolicy::Fixed {
                machines: existing_machines,
            },
            CapacityGroupScalePolicy::Fixed {
                machines: declared_machines,
            },
        ) => CapacityGroupScalePolicy::Fixed {
            machines: merge_choice_range(existing_machines, declared_machines),
        },
        (_, declared) => declared.clone(),
    }
}

fn merge_choice_range(
    existing: &ComputeChoiceRange,
    declared: &ComputeChoiceRange,
) -> ComputeChoiceRange {
    ComputeChoiceRange {
        min: existing.min.max(declared.min),
        max: existing.max.max(declared.max),
        default: declared.default,
    }
}

fn validate_selection_against_scale(
    selection: &ComputePoolSelection,
    scale: &CapacityGroupScalePolicy,
) -> std::result::Result<(), String> {
    // A fixed pool of N machines and an autoscaling pool bounded to N..N run the same
    // fleet, so either form is accepted wherever the other one is.
    match (selection, scale) {
        (
            ComputePoolSelection::Fixed { machines, .. },
            CapacityGroupScalePolicy::Fixed { machines: allowed },
        ) => check_machine_count("fixed machine count", *machines, allowed.min, allowed.max),
        (
            ComputePoolSelection::Autoscale { min, max, .. },
            CapacityGroupScalePolicy::Autoscale {
                min: allowed_min,
                max: allowed_max,
            },
        ) => {
            check_machine_count("autoscale minimum", *min, allowed_min.min, allowed_min.max)?;
            check_machine_count("autoscale maximum", *max, allowed_max.min, allowed_max.max)
        }
        (
            ComputePoolSelection::Fixed { machines, .. },
            CapacityGroupScalePolicy::Autoscale {
                min: allowed_min,
                max: allowed_max,
            },
        ) => {
            let lowest = allowed_min.min.max(allowed_max.min);
            let highest = allowed_min.max.min(allowed_max.max);
            if lowest > highest {
                return Err(format!(
                    "must use autoscale mode (minimum {}-{}, maximum {}-{})",
                    allowed_min.min, allowed_min.max, allowed_max.min, allowed_max.max
                ));
            }
            check_machine_count("fixed machine count", *machines, lowest, highest)
        }
        (
            ComputePoolSelection::Autoscale { min, max, .. },
            CapacityGroupScalePolicy::Fixed { machines: allowed },
        ) => {
            if min != max {
                return Err(format!(
                    "must use fixed mode with {}-{} machines; autoscale {min}-{max} is not allowed",
                    allowed.min, allowed.max
                ));
            }
            check_machine_count("machine count", *min, allowed.min, allowed.max)
        }
    }
}

fn check_machine_count(
    label: &str,
    value: u32,
    lowest: u32,
    highest: u32,
) -> std::result::Result<(), String> {
    if (lowest..=highest).contains(&value) {
        Ok(())
    } else {
        Err(format!(
            "{label} {value} is outside the allowed range {lowest}-{highest}"
        ))
    }
}

fn needed_container_pool(container: &Container) -> &'static str {
    if container.stateful && container.persistent_storage.is_some() {
        return "stateful";
    }
    if container.gpu.is_some() {
        return "gpu";
    }
    if let Some(storage) = &container.ephemeral_storage {
        if instance_catalog::parse_memory_bytes(storage).unwrap_or(0) > 200 * 1024 * 1024 * 1024 {
            return "storage";
        }
    }
    "general"
}

fn default_min_machines(requirements: &WorkloadRequirements) -> u32 {
    if requirements.total_cpu_at_desired > 0.0 || requirements.total_memory_bytes_at_desired > 0 {
        1
    } else {
        0
    }
}

fn default_max_machines(requirements: &WorkloadRequirements) -> u32 {
    let min = default_min_machines(requirements);
    let by_cpu =
        (requirements.total_cpu_at_max / requirements.max_cpu_per_container.max(1.0)).ceil() as u32;
    let by_mem = requirements
        .total_memory_bytes_at_max
        .div_ceil(requirements.max_memory_per_container.max(1)) as u32;
    min.max(by_cpu).max(by_mem).max(1)
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{
        instance_catalog::Architecture, CapacityGroup, CapacityGroupScalePolicy,
        ComputeChoiceRange, ComputeCluster, ComputeSettings, ContainerCode, DaemonCode, Resource,
        ResourceEntry, ResourceLifecycle, Stack, Storage, Worker, WorkerCode,
    };

    #[test]
    fn deployment_resources_drive_machine_planning_on_every_cloud() {
        let mut stack = stack_with_container();
        let container = stack
            .resources
            .get_mut("api")
            .unwrap()
            .config
            .downcast_mut::<Container>()
            .unwrap();
        container.resource_choices = Some(crate::ContainerResourceChoices {
            cpu: Some(crate::ResourceChoiceRange {
                min: "0.5".into(),
                max: "8".into(),
                default: "2".into(),
            }),
            memory: Some(crate::ResourceChoiceRange {
                min: "512Mi".into(),
                max: "16Gi".into(),
                default: "4Gi".into(),
            }),
        });
        let settings: ComputeSettings = serde_json::from_value(serde_json::json!({
            "pools": { "general": { "mode": "fixed", "machines": 1 } },
            "containers": { "api": { "cpu": 8, "memory": "16Gi" } }
        }))
        .unwrap();
        for platform in [Platform::Aws, Platform::Gcp, Platform::Azure] {
            let default_plan = plan_compute(&stack, platform, None).unwrap();
            let selected_plan = plan_compute(&stack, platform, Some(&settings)).unwrap();
            let pool = &selected_plan.pools[0];
            assert!(pool.errors.is_empty(), "{platform}: {:?}", pool.errors);
            assert_eq!(pool.requirements.cpu, "8");
            assert_eq!(pool.requirements.memory_bytes, 16 * 1024 * 1024 * 1024);
            assert_ne!(
                pool.selected.machine(),
                default_plan.pools[0].selected.machine()
            );
            assert_eq!(selected_plan.containers[0].cpu.min, "8");
            assert_eq!(selected_plan.containers[0].memory.desired, "16Gi");
            assert_eq!(pool.selected.min_size(), 1);
            assert_eq!(pool.selected.max_size(), 1);
        }
        assert_eq!(
            settings.pools["general"].machine(),
            None,
            "resolving a plan must preserve automatic machine intent"
        );
        let original = stack.resources["api"]
            .config
            .downcast_ref::<Container>()
            .unwrap();
        assert_eq!(
            original.cpu.desired, "2",
            "planning must not mutate release defaults"
        );
    }

    #[test]
    fn explicit_machine_can_fit_a_low_cpu_high_memory_workload() {
        let mut stack = stack_with_container();
        let container = stack
            .resources
            .get_mut("api")
            .unwrap()
            .config
            .downcast_mut::<Container>()
            .unwrap();
        container.cpu = ResourceSpec {
            min: "0.5".into(),
            desired: "0.5".into(),
        };
        container.memory = ResourceSpec {
            min: "8Gi".into(),
            desired: "8Gi".into(),
        };
        let settings: ComputeSettings = serde_json::from_value(serde_json::json!({
            "pools": {"general":{"mode":"fixed","machines":1,"machine":"n2-standard-4"}}
        }))
        .unwrap();
        let plan = plan_compute(&stack, Platform::Gcp, Some(&settings)).unwrap();
        assert!(
            plan.pools[0].errors.is_empty(),
            "{:?}",
            plan.pools[0].errors
        );
        assert_eq!(plan.pools[0].selected.machine(), Some("n2-standard-4"));
    }

    #[test]
    fn pin_must_leave_host_capacity_for_one_replica() {
        let mut stack = stack_with_container();
        stack
            .resources
            .get_mut("api")
            .unwrap()
            .config
            .downcast_mut::<Container>()
            .unwrap()
            .cpu = ResourceSpec {
            min: "4".into(),
            desired: "4".into(),
        };
        let settings: ComputeSettings = serde_json::from_value(serde_json::json!({
            "pools":{"general":{"mode":"fixed","machines":1,"machine":"t3.xlarge"}}
        }))
        .unwrap();
        let plan = plan_compute(&stack, Platform::Aws, Some(&settings)).unwrap();
        assert!(
            plan.pools[0]
                .errors
                .iter()
                .any(|error| error.contains("host resources")),
            "{:?}",
            plan.pools[0].errors
        );
    }

    #[test]
    fn valid_pin_does_not_require_automatic_recommendation_headroom() {
        let machine = instance_catalog::catalog_for_platform(Platform::Gcp)
            .into_iter()
            .max_by_key(|machine| machine.memory_bytes)
            .unwrap();
        let mut stack = stack_with_container();
        let container = stack
            .resources
            .get_mut("api")
            .unwrap()
            .config
            .downcast_mut::<Container>()
            .unwrap();
        container.cpu = ResourceSpec {
            min: "0.5".into(),
            desired: "0.5".into(),
        };
        let memory = instance_catalog::allocatable_memory_bytes(machine).to_string();
        container.memory = ResourceSpec {
            min: memory.clone(),
            desired: memory,
        };
        assert!(plan_compute(&stack, Platform::Gcp, None).is_err());
        let settings: ComputeSettings = serde_json::from_value(serde_json::json!({
            "pools":{"general":{"mode":"fixed","machines":1,"machine":machine.name}}
        }))
        .unwrap();
        let plan = plan_compute(&stack, Platform::Gcp, Some(&settings)).unwrap();
        assert!(
            plan.pools[0].errors.is_empty(),
            "{:?}",
            plan.pools[0].errors
        );
        assert_eq!(plan.pools[0].selected.machine(), Some(machine.name));
    }

    #[test]
    fn invalid_deployment_resources_fail_before_capacity_planning() {
        let mut stack = stack_with_container();
        stack
            .resources
            .get_mut("api")
            .unwrap()
            .config
            .downcast_mut::<Container>()
            .unwrap()
            .resource_choices = Some(crate::ContainerResourceChoices {
            cpu: Some(crate::ResourceChoiceRange {
                min: "0.5".into(),
                max: "4".into(),
                default: "2".into(),
            }),
            memory: Some(crate::ResourceChoiceRange {
                min: "512Mi".into(),
                max: "8Gi".into(),
                default: "4Gi".into(),
            }),
        });
        for selection in [
            serde_json::json!({"api":{"cpu": 0}}),
            serde_json::json!({"api":{"cpu": 5}}),
            serde_json::json!({"api":{"memory": "16Gi"}}),
            serde_json::json!({"api":{"memory": "NaNGi"}}),
            serde_json::json!({"api":{"memory": "-1Gi"}}),
            serde_json::json!({"missing":{"cpu": 1}}),
        ] {
            let settings: ComputeSettings =
                serde_json::from_value(serde_json::json!({"containers": selection})).unwrap();
            let error = plan_compute(&stack, Platform::Aws, Some(&settings)).unwrap_err();
            assert_eq!(error.code, "CONTAINER_RESOURCE_SELECTION_INVALID");
        }
    }

    #[test]
    fn deployment_volume_selection_is_applied_without_changing_release_defaults() {
        let mut release = stack_with_container();
        release
            .resources
            .get_mut("api")
            .unwrap()
            .config
            .downcast_mut::<Container>()
            .unwrap()
            .persistent_storage = Some(crate::PersistentStorage {
            size: "20Gi".into(),
            mount_path: "/data".into(),
            backups: crate::VolumeBackups::default(),
            encryption_key: None,
        });
        let settings: ComputeSettings = serde_json::from_value(serde_json::json!({
            "containers": {"api": {"volumeSizeGib": 40}}
        }))
        .unwrap();
        let mut deployment = release.clone();
        crate::container_resources::resolve_container_resources(&mut deployment, Some(&settings))
            .unwrap();
        let volume = |stack: &Stack| {
            stack.resources["api"]
                .config
                .downcast_ref::<Container>()
                .unwrap()
                .persistent_storage
                .as_ref()
                .unwrap()
                .size
                .clone()
        };
        assert_eq!(volume(&deployment), "40Gi");
        assert_eq!(volume(&release), "20Gi");
        let mut equivalent = release.clone();
        equivalent
            .resources
            .get_mut("api")
            .unwrap()
            .config
            .downcast_mut::<Container>()
            .unwrap()
            .persistent_storage
            .as_mut()
            .unwrap()
            .size = "1Ti".into();
        let same_capacity: ComputeSettings = serde_json::from_value(serde_json::json!({
            "containers": {"api": {"volumeSizeGib": 1024}}
        }))
        .unwrap();
        crate::container_resources::resolve_container_resources(
            &mut equivalent,
            Some(&same_capacity),
        )
        .unwrap();
        assert_eq!(
            volume(&equivalent),
            "1Ti",
            "equivalent selections preserve PVC quantities"
        );
        crate::container_resources::resolve_container_resources(&mut deployment, Some(&settings))
            .unwrap();
        assert_eq!(volume(&deployment), "40Gi");
        let smaller_initial: ComputeSettings = serde_json::from_value(serde_json::json!({
            "containers": {"api": {"volumeSizeGib": 10}}
        }))
        .unwrap();
        crate::container_resources::resolve_container_resources(
            &mut release.clone(),
            Some(&smaller_initial),
        )
        .unwrap();
        let mut installed = StackState::new(Platform::Aws);
        installed.resources.insert(
            "api".into(),
            crate::StackResourceState::new_pending(
                "Container".into(),
                deployment.resources["api"].config.clone(),
                None,
                vec![],
            ),
        );
        let shrinking: ComputeSettings = serde_json::from_value(serde_json::json!({
            "containers": {"api": {"volumeSizeGib": 30}}
        }))
        .unwrap();
        let error =
            plan_compute_with_state(&release, Platform::Aws, Some(&shrinking), Some(&installed))
                .unwrap_err();
        assert_eq!(error.code, "CONTAINER_RESOURCE_SELECTION_INVALID");
        assert!(error.to_string().contains("cannot shrink from 40 to 30"));
        let imported: ComputeSettings = serde_json::from_value(serde_json::json!({
            "containers": {"api": {"volumeSizeGib": "40"}}
        }))
        .unwrap();
        assert_eq!(imported.containers["api"].volume_size_gib, Some(40));
        for value in [
            serde_json::json!(-1),
            serde_json::json!(1.5),
            serde_json::json!("4294967296"),
        ] {
            assert!(
                serde_json::from_value::<ComputeSettings>(serde_json::json!({
                    "containers": {"api": {"volumeSizeGib": value}}
                }))
                .is_err()
            );
        }
        for (platform, size) in [(Platform::Gcp, 9), (Platform::Azure, 32768)] {
            let invalid: ComputeSettings = serde_json::from_value(serde_json::json!({
                "containers": {"api": {"volumeSizeGib": size}}
            }))
            .unwrap();
            let error = plan_compute(&release, platform, Some(&invalid)).unwrap_err();
            assert_eq!(error.code, "CONTAINER_RESOURCE_SELECTION_INVALID");
            assert!(error.to_string().contains("must be between"));
        }
        for size in [0, 65537] {
            let invalid: ComputeSettings = serde_json::from_value(serde_json::json!({
                "containers": {"api": {"volumeSizeGib": size}}
            }))
            .unwrap();
            assert_eq!(
                crate::container_resources::resolve_container_resources(
                    &mut release.clone(),
                    Some(&invalid)
                )
                .unwrap_err()
                .code,
                "CONTAINER_RESOURCE_SELECTION_INVALID"
            );
        }
        assert_eq!(
            crate::container_resources::resolve_container_resources(
                &mut stack_with_container(),
                Some(&settings)
            )
            .unwrap_err()
            .code,
            "CONTAINER_RESOURCE_SELECTION_INVALID"
        );
    }

    fn stack_with_volume(size: &str) -> Stack {
        let mut stack = stack_with_container();
        stack
            .resources
            .get_mut("api")
            .unwrap()
            .config
            .downcast_mut::<Container>()
            .unwrap()
            .persistent_storage = Some(crate::PersistentStorage {
            size: size.into(),
            mount_path: "/data".into(),
            backups: crate::VolumeBackups::default(),
            encryption_key: None,
        });
        stack
    }

    fn installed_with_volume(platform: Platform, size: &str) -> StackState {
        let mut installed = StackState::new(platform);
        installed.resources.insert(
            "api".into(),
            crate::StackResourceState::new_pending(
                "Container".into(),
                stack_with_volume(size).resources["api"].config.clone(),
                None,
                vec![],
            ),
        );
        installed
    }

    fn api_volume(plan: &ComputePlan) -> (Option<u32>, Option<u32>) {
        let api = plan
            .containers
            .iter()
            .find(|container| container.container_id == "api")
            .expect("api is planned");
        (api.volume_size_gib, api.minimum_volume_size_gib)
    }

    /// A release that lowers its default volume size must not turn into a shrink of every
    /// installed deployment that never chose a size: they keep their installed volumes. A
    /// larger default still grows them.
    #[test]
    fn a_lower_release_default_keeps_installed_volumes_without_a_selection() {
        for platform in [Platform::Aws, Platform::Gcp, Platform::Azure] {
            let installed = installed_with_volume(platform, "30Gi");
            let plan = plan_compute_with_state(
                &stack_with_volume("20Gi"),
                platform,
                None,
                Some(&installed),
            )
            .expect("a lower release default keeps the installed volume");
            assert_eq!(api_volume(&plan), (Some(30), Some(30)), "{platform}");

            let plan = plan_compute_with_state(
                &stack_with_volume("40Gi"),
                platform,
                None,
                Some(&installed),
            )
            .expect("a larger release default grows the volume");
            assert_eq!(api_volume(&plan), (Some(40), Some(30)), "{platform}");

            // A selection is the deployment's own choice, so it still cannot shrink.
            let shrinking: ComputeSettings = serde_json::from_value(serde_json::json!({
                "containers": {"api": {"volumeSizeGib": 25}}
            }))
            .unwrap();
            let error = plan_compute_with_state(
                &stack_with_volume("20Gi"),
                platform,
                Some(&shrinking),
                Some(&installed),
            )
            .unwrap_err();
            assert!(
                error.to_string().contains("cannot shrink from 30 to 25"),
                "{error}"
            );
        }
    }

    /// Setup sources (CloudFormation parameters, Terraform variables, CLI config files) cannot see
    /// growth made through a compute update. Their lower values keep the installed size; equal
    /// or larger values, and containers without an installed volume, are left alone.
    #[test]
    fn setup_source_volume_sizes_below_installed_keep_the_installed_size() {
        let mut installed = installed_with_volume(Platform::Aws, "30Gi");
        let mut worker = stack_with_volume("10Gi").resources["api"].config.clone();
        worker
            .downcast_mut::<Container>()
            .unwrap()
            .persistent_storage = None;
        installed.resources.insert(
            "worker".into(),
            crate::StackResourceState::new_pending("Container".into(), worker, None, vec![]),
        );
        let mut settings: ComputeSettings = serde_json::from_value(serde_json::json!({
            "containers": {
                "api": {"volumeSizeGib": 25, "cpu": 2},
                "worker": {"volumeSizeGib": 5},
                "new": {"volumeSizeGib": 8}
            }
        }))
        .unwrap();

        let kept = crate::container_resources::keep_volume_sizes_over_setup_source(
            &mut settings,
            Some(&installed),
            None,
        );

        assert_eq!(
            kept,
            vec![crate::container_resources::KeptVolumeSize {
                container_id: "api".into(),
                requested_gib: 25,
                kept_gib: 30,
            }]
        );
        assert_eq!(settings.containers["api"].volume_size_gib, Some(30));
        assert_eq!(
            settings.containers["api"].cpu,
            Some(serde_json::Number::from(2))
        );
        assert_eq!(settings.containers["worker"].volume_size_gib, Some(5));
        assert_eq!(settings.containers["new"].volume_size_gib, Some(8));
        let plan = plan_compute_with_state(
            &stack_with_volume("20Gi"),
            Platform::Aws,
            Some(&ComputeSettings {
                containers: [("api".to_string(), settings.containers["api"].clone())].into(),
                ..Default::default()
            }),
            Some(&installed_with_volume(Platform::Aws, "30Gi")),
        )
        .expect("the kept size plans");
        assert_eq!(api_volume(&plan), (Some(30), Some(30)));

        let mut growing: ComputeSettings = serde_json::from_value(serde_json::json!({
            "containers": {"api": {"volumeSizeGib": 35}}
        }))
        .unwrap();
        assert!(
            crate::container_resources::keep_volume_sizes_over_setup_source(
                &mut growing,
                Some(&installed),
                None,
            )
            .is_empty()
        );
        assert_eq!(growing.containers["api"].volume_size_gib, Some(35));

        // A growth the deployment chose but has not applied yet is not undone either.
        let chosen: ComputeSettings = serde_json::from_value(serde_json::json!({
            "containers": {"api": {"volumeSizeGib": 40}}
        }))
        .unwrap();
        let mut stale: ComputeSettings = serde_json::from_value(serde_json::json!({
            "containers": {"api": {"volumeSizeGib": 25}}
        }))
        .unwrap();
        let kept = crate::container_resources::keep_volume_sizes_over_setup_source(
            &mut stale,
            Some(&installed),
            Some(&chosen),
        );
        assert_eq!(kept[0].kept_gib, 40);
        assert_eq!(stale.containers["api"].volume_size_gib, Some(40));

        // Before anything is installed (a first install whose preflights failed), a smaller
        // value is a correction of the saved choice and is kept as asked.
        let mut correction: ComputeSettings = serde_json::from_value(serde_json::json!({
            "containers": {"api": {"volumeSizeGib": 20}}
        }))
        .unwrap();
        for nothing_installed in [None, Some(&StackState::new(Platform::Aws))] {
            assert!(
                crate::container_resources::keep_volume_sizes_over_setup_source(
                    &mut correction,
                    nothing_installed,
                    Some(&chosen),
                )
                .is_empty()
            );
            assert_eq!(correction.containers["api"].volume_size_gib, Some(20));
        }
    }

    #[test]
    fn installed_storage_cannot_be_removed_or_remounted_on_any_provider() {
        for platform in [
            Platform::Aws,
            Platform::Gcp,
            Platform::Azure,
            Platform::Kubernetes,
        ] {
            let mut release = stack_with_container();
            release
                .resources
                .get_mut("api")
                .unwrap()
                .config
                .downcast_mut::<Container>()
                .unwrap()
                .persistent_storage = Some(crate::PersistentStorage {
                size: "20Gi".into(),
                mount_path: "/data".into(),
                backups: crate::VolumeBackups::default(),
                encryption_key: None,
            });
            let mut installed = StackState::new(platform);
            installed.resources.insert(
                "api".into(),
                crate::StackResourceState::new_pending(
                    "Container".into(),
                    release.resources["api"].config.clone(),
                    None,
                    vec![],
                ),
            );
            for remove in [false, true] {
                let mut desired = release.clone();
                let storage = &mut desired
                    .resources
                    .get_mut("api")
                    .unwrap()
                    .config
                    .downcast_mut::<Container>()
                    .unwrap()
                    .persistent_storage;
                if remove {
                    *storage = None;
                } else {
                    storage.as_mut().unwrap().mount_path = "/other".into();
                }
                let error = plan_compute_with_state(&desired, platform, None, Some(&installed))
                    .unwrap_err();
                assert_eq!(error.code, "CONTAINER_RESOURCE_SELECTION_INVALID");
                assert!(
                    error.to_string().contains(if remove {
                        "removing installed"
                    } else {
                        "mount path"
                    }),
                    "{error}"
                );
            }
        }
    }

    #[test]
    fn release_volume_defaults_obey_provider_bounds_without_a_selection() {
        for (platform, size) in [(Platform::Gcp, "5Gi"), (Platform::Azure, "40000Gi")] {
            let mut stack = stack_with_container();
            stack
                .resources
                .get_mut("api")
                .unwrap()
                .config
                .downcast_mut::<Container>()
                .unwrap()
                .persistent_storage = Some(crate::PersistentStorage {
                size: size.into(),
                mount_path: "/data".into(),
                backups: crate::VolumeBackups::default(),
                encryption_key: None,
            });
            let error = plan_compute(&stack, platform, None).unwrap_err();
            assert_eq!(error.code, "CONTAINER_RESOURCE_SELECTION_INVALID");
            assert!(error.to_string().contains("must be between"));
        }
    }

    #[test]
    fn volume_capacity_parsing_rejects_overflow_and_preserves_legacy_units() {
        use crate::container_resources::parse_volume_size_gib;
        for (value, size) in [
            ("20Gi", 20),
            ("20GiB", 20),
            ("20GB", 20),
            ("1Ti", 1024),
            ("1TB", 1000),
            ("1.5Ti", 1536),
            ("1536Mi", 2),
            ("0.5Gi", 1),
            ("1073741824", 1),
        ] {
            assert_eq!(parse_volume_size_gib(value).unwrap(), size);
        }
        for value in ["0Gi", "-1Gi", "NaNGi", "4294967295Ti", "65537Gi"] {
            assert!(parse_volume_size_gib(value).is_err(), "{value}");
        }
    }

    fn stack_with_container() -> Stack {
        let container = Container::new("api".to_string())
            .code(ContainerCode::Image {
                image: "api:latest".to_string(),
            })
            .cpu(ResourceSpec {
                min: "1".to_string(),
                desired: "2".to_string(),
            })
            .memory(ResourceSpec {
                min: "2Gi".to_string(),
                desired: "4Gi".to_string(),
            })
            .permissions("api".to_string())
            .build();
        Stack {
            id: "test".to_string(),
            resources: [(
                "api".to_string(),
                ResourceEntry {
                    config: Resource::new(container),
                    lifecycle: ResourceLifecycle::Live,
                    dependencies: Vec::new(),
                    display_name: None,
                    description: None,
                    remote_access: false,
                    enabled_when: None,
                },
            )]
            .into_iter()
            .collect(),
            permissions: crate::permissions::PermissionsConfig::default(),
            supported_platforms: None,
            inputs: vec![],
            dynamic_container_repositories: Vec::new(),
            dynamic_container_image_resources: Vec::new(),
            operations: None,
        }
    }

    #[test]
    fn serverless_workers_and_storage_do_not_allocate_machine_pools() {
        let mut stack = stack_with_container();
        stack.resources.clear();
        for (id, resource) in [
            (
                "api",
                Resource::new(
                    Worker::new("api".to_string())
                        .code(WorkerCode::Image {
                            image: "example/worker:v1".to_string(),
                        })
                        .permissions("execution".to_string())
                        .build(),
                ),
            ),
            (
                "data",
                Resource::new(Storage::new("data".to_string()).build()),
            ),
        ] {
            stack.resources.insert(
                id.to_string(),
                ResourceEntry {
                    config: resource,
                    lifecycle: ResourceLifecycle::Frozen,
                    dependencies: Vec::new(),
                    display_name: None,
                    description: None,
                    remote_access: false,
                    enabled_when: None,
                },
            );
        }

        for platform in [Platform::Aws, Platform::Gcp, Platform::Azure] {
            let plan = plan_compute(&stack, platform, None).expect("plan should build");
            assert!(plan.pools.is_empty(), "serverless plan allocated {plan:?}");
        }
    }

    #[test]
    fn empty_stack_does_not_allocate_machine_pools() {
        let mut stack = stack_with_container();
        stack.resources.clear();
        let plan = plan_compute(&stack, Platform::Aws, None).expect("plan should build");
        assert!(plan.pools.is_empty());
    }

    #[test]
    fn explicit_cluster_without_workloads_only_plans_declared_pools() {
        let mut stack = stack_with_container();
        stack.resources.clear();
        let cluster = ComputeCluster::new("compute".to_string())
            .capacity_group(CapacityGroup {
                group_id: "batch".to_string(),
                instance_type: Some("t4g.medium".to_string()),
                profile: None,
                min_size: 1,
                max_size: 1,
                scale_policy: None,
                nested_virtualization: None,
                data_disk_gib_range: None,
            })
            .build();
        stack.resources.insert(
            "compute".to_string(),
            ResourceEntry {
                config: Resource::new(cluster),
                lifecycle: ResourceLifecycle::Frozen,
                dependencies: Vec::new(),
                display_name: None,
                description: None,
                remote_access: false,
                enabled_when: None,
            },
        );

        let plan = plan_compute(&stack, Platform::Aws, None).expect("plan should build");
        assert_eq!(plan.pools.len(), 1);
        assert_eq!(plan.pools[0].pool_id, "batch");
        assert!(plan.pools[0].errors.is_empty());
    }

    #[test]
    fn cloud_plan_recommends_provider_machine_without_mutating_selection() {
        let stack = stack_with_container();

        let plan = plan_compute(&stack, Platform::Aws, None).expect("plan should build");

        let pool = plan.pools.first().expect("general pool should exist");
        assert_eq!(pool.pool_id, "general");
        assert_eq!(pool.workloads, vec!["api"]);
        assert!(pool.selected.machine().is_some());
        assert!(pool.machines.iter().any(|machine| machine.recommended));
    }

    #[test]
    fn cloud_plan_only_offers_machines_matching_the_default_image_target() {
        let stack = stack_with_container();

        for platform in [Platform::Aws, Platform::Gcp, Platform::Azure] {
            let plan = plan_compute(&stack, platform, None).expect("plan should build");
            let pool = plan.pools.first().expect("general pool should exist");
            let expected = instance_catalog::default_architecture(platform)
                .expect("managed cloud should have a default architecture");

            assert!(pool.machines.iter().all(|machine| {
                instance_catalog::find_instance_type(platform, &machine.machine)
                    .is_some_and(|spec| spec.architecture == expected)
            }));
        }
    }

    #[test]
    fn selected_machine_is_preserved_as_static_deployment_choice() {
        let stack = stack_with_container();
        let settings = ComputeSettings {
            containers: Default::default(),
            pools: [(
                "general".to_string(),
                ComputePoolSelection::Fixed {
                    machines: 1,
                    machine: Some("m7g.xlarge".to_string()),
                    failure_domains: None,
                    data_disk_gib: None,
                },
            )]
            .into_iter()
            .collect(),
        };

        let plan = plan_compute(&stack, Platform::Aws, Some(&settings)).expect("plan should build");

        let pool = plan.pools.first().expect("general pool should exist");
        assert_eq!(pool.selected.machine(), Some("m7g.xlarge"));
        assert!(pool.errors.is_empty());
    }

    #[test]
    fn configurable_cloud_disk_satisfies_storage_above_fixed_local_catalog() {
        let mut stack = stack_with_container();
        let container = stack
            .resources
            .get_mut("api")
            .and_then(|entry| entry.config.downcast_mut::<Container>())
            .expect("test stack should contain a container");
        container.ephemeral_storage = Some("8000Gi".to_string());
        let settings = ComputeSettings {
            containers: Default::default(),
            pools: [(
                "storage".to_string(),
                ComputePoolSelection::Fixed {
                    machines: 1,
                    machine: Some("m8i.2xlarge".to_string()),
                    failure_domains: None,
                    data_disk_gib: None,
                },
            )]
            .into_iter()
            .collect(),
        };

        let plan = plan_compute(&stack, Platform::Aws, Some(&settings)).expect("plan should build");
        let pool = plan.pools.first().expect("general pool should exist");
        assert!(pool.errors.is_empty());
        let selected = pool
            .machines
            .iter()
            .find(|option| option.machine == "m8i.2xlarge")
            .expect("selected machine should remain an option");
        assert_eq!(
            selected.profile.ephemeral_storage_bytes,
            8_000 * 1024 * 1024 * 1024
        );
    }

    #[test]
    fn selected_machine_defines_architecture_when_workloads_do_not() {
        let stack = stack_with_container();
        let settings = ComputeSettings {
            containers: Default::default(),
            pools: [(
                "general".to_string(),
                ComputePoolSelection::Fixed {
                    machines: 1,
                    machine: Some("m7i.xlarge".to_string()),
                    failure_domains: None,
                    data_disk_gib: None,
                },
            )]
            .into_iter()
            .collect(),
        };

        let plan = plan_compute(&stack, Platform::Aws, Some(&settings)).expect("plan should build");

        let pool = plan.pools.first().expect("general pool should exist");
        assert!(pool.errors.is_empty());
        assert!(pool.machines.iter().all(|machine| {
            instance_catalog::find_instance_type(Platform::Aws, &machine.machine)
                .is_some_and(|spec| spec.architecture == Architecture::X86_64)
        }));
    }

    #[test]
    fn graviton4_compute_machine_can_be_selected() {
        let stack = stack_with_container();
        let settings = ComputeSettings {
            containers: Default::default(),
            pools: [(
                "general".to_string(),
                ComputePoolSelection::Fixed {
                    machines: 1,
                    machine: Some("c8g.2xlarge".to_string()),
                    failure_domains: None,
                    data_disk_gib: None,
                },
            )]
            .into_iter()
            .collect(),
        };

        let plan = plan_compute(&stack, Platform::Aws, Some(&settings)).expect("plan should build");

        let pool = plan.pools.first().expect("general pool should exist");
        assert!(pool.errors.is_empty(), "{:?}", pool.errors);
        let selected = pool
            .machines
            .iter()
            .find(|option| option.machine == "c8g.2xlarge")
            .expect("selected machine should be offered");
        assert!(selected.recommended);
        assert_eq!(selected.profile.architecture, Some(Architecture::Arm64));
    }

    #[test]
    fn explicit_capacity_group_requirements_are_merged_with_workloads() {
        let mut stack = stack_with_container();
        let cluster = ComputeCluster::new("compute".to_string())
            .capacity_group(CapacityGroup {
                group_id: "general".to_string(),
                instance_type: None,
                profile: Some(MachineProfile {
                    cpu: "4".to_string(),
                    memory_bytes: 16 * 1024 * 1024 * 1024,
                    ephemeral_storage_bytes: 20 * 1024 * 1024 * 1024,
                    architecture: Some(Architecture::X86_64),
                    gpu: None,
                    data_disk: None,
                }),
                min_size: 2,
                max_size: 5,
                scale_policy: None,
                nested_virtualization: Some(true),
                data_disk_gib_range: None,
            })
            .build();
        stack.resources.insert(
            "compute".to_string(),
            ResourceEntry {
                config: Resource::new(cluster),
                lifecycle: ResourceLifecycle::Frozen,
                dependencies: Vec::new(),
                display_name: None,
                description: None,
                remote_access: false,
                enabled_when: None,
            },
        );

        let plan = plan_compute(&stack, Platform::Aws, None).expect("plan should build");

        let pool = plan.pools.first().expect("general pool should exist");
        let machine = pool
            .selected
            .machine()
            .expect("AWS selection should include a machine");
        let spec = instance_catalog::find_instance_type(Platform::Aws, machine)
            .expect("selected machine should exist in the catalog");
        assert!(spec.is_nested_virt_capable());
        assert_eq!(pool.selected.min_size(), 2);
        assert_eq!(pool.selected.max_size(), 5);
        assert!(pool.errors.is_empty());
    }

    /// The pool path deployment packages plan: a nested-virtualization pool that
    /// leaves the architecture unset must resolve to the x86_64 machines AWS
    /// offers for it, the same machine the image build resolves.
    #[test]
    fn nested_pool_without_architecture_plans_an_aws_machine() {
        let mut stack = stack_with_container();
        let cluster = ComputeCluster::new("compute".to_string())
            .capacity_group(CapacityGroup {
                group_id: "general".to_string(),
                instance_type: None,
                profile: Some(MachineProfile {
                    cpu: "4".to_string(),
                    memory_bytes: 16 * 1024 * 1024 * 1024,
                    ephemeral_storage_bytes: 20 * 1024 * 1024 * 1024,
                    architecture: None,
                    gpu: None,
                    data_disk: None,
                }),
                min_size: 1,
                max_size: 1,
                scale_policy: None,
                nested_virtualization: Some(true),
                data_disk_gib_range: None,
            })
            .build();
        stack.resources.insert(
            "compute".to_string(),
            ResourceEntry {
                config: Resource::new(cluster),
                lifecycle: ResourceLifecycle::Frozen,
                dependencies: Vec::new(),
                display_name: None,
                description: None,
                remote_access: false,
                enabled_when: None,
            },
        );

        let plan = plan_compute(&stack, Platform::Aws, None).expect("plan should build");

        let pool = plan.pools.first().expect("general pool should exist");
        assert!(pool.errors.is_empty(), "{:?}", pool.errors);
        let machine = pool
            .selected
            .machine()
            .expect("AWS selection should include a machine");
        let spec = instance_catalog::find_instance_type(Platform::Aws, machine)
            .expect("selected machine should exist in the catalog");
        assert!(spec.is_nested_virt_capable(), "{machine} is not nested-capable");
        assert_eq!(spec.architecture, Architecture::X86_64);
        assert!(!pool.machines.is_empty());
        assert!(pool.machines.iter().all(|option| {
            instance_catalog::find_instance_type(Platform::Aws, &option.machine).is_some_and(
                |spec| spec.is_nested_virt_capable() && spec.architecture == Architecture::X86_64,
            )
        }));
    }

    const GIB: u64 = 1024 * 1024 * 1024;

    /// A stack with one declared `general` pool. `data_disk_gib` is the
    /// declared size (the default for a range) and `range` the sizes offered.
    fn stack_with_data_disk_pool(
        data_disk_gib: Option<u64>,
        range: Option<ComputeChoiceRange>,
    ) -> Stack {
        let mut stack = stack_with_container();
        let cluster = ComputeCluster::new("compute".to_string())
            .capacity_group(CapacityGroup {
                group_id: "general".to_string(),
                instance_type: None,
                profile: Some(MachineProfile {
                    cpu: "2".to_string(),
                    memory_bytes: 8 * GIB,
                    ephemeral_storage_bytes: 20 * GIB,
                    architecture: None,
                    gpu: None,
                    data_disk: data_disk_gib
                        .map(|gib| crate::DataDiskRequirement { size_bytes: gib * GIB }),
                }),
                min_size: 1,
                max_size: 1,
                scale_policy: None,
                nested_virtualization: None,
                data_disk_gib_range: range,
            })
            .build();
        stack.resources.insert(
            "compute".to_string(),
            ResourceEntry {
                config: Resource::new(cluster),
                lifecycle: ResourceLifecycle::Frozen,
                dependencies: Vec::new(),
                display_name: None,
                description: None,
                remote_access: false,
                enabled_when: None,
            },
        );
        stack
    }

    fn data_disk_selection(data_disk_gib: Option<u32>) -> ComputeSettings {
        ComputeSettings {
            pools: [(
                "general".to_string(),
                ComputePoolSelection::Fixed {
                    machines: 1,
                    machine: None,
                    failure_domains: None,
                    data_disk_gib,
                },
            )]
            .into_iter()
            .collect(),
            containers: HashMap::new(),
        }
    }

    fn general_pool(plan: &ComputePlan) -> &ComputePoolPlan {
        plan.pools
            .iter()
            .find(|pool| pool.pool_id == "general")
            .expect("general pool should be planned")
    }

    #[test]
    fn a_deployment_chooses_its_data_disk_within_the_declared_range() {
        let range = ComputeChoiceRange {
            min: 0,
            max: 4096,
            default: 200,
        };
        let stack = stack_with_data_disk_pool(Some(200), Some(range));
        for platform in [Platform::Aws, Platform::Machines] {
            for (selected, expected_gib) in [(None, 200), (Some(50), 50), (Some(0), 0)] {
                let settings = data_disk_selection(selected);
                let plan = plan_compute(&stack, platform, Some(&settings)).expect("plan");
                let pool = general_pool(&plan);
                assert!(pool.errors.is_empty(), "{platform} {selected:?}: {:?}", pool.errors);
                assert_eq!(
                    pool.data_disk,
                    Some(ComputePoolDataDiskPlan {
                        min_gib: 0,
                        max_gib: 4096,
                        default_gib: 200,
                        selected_gib: expected_gib,
                    }),
                    "{platform} {selected:?}"
                );
                // An omitted choice stays omitted, so later releases can move the default.
                assert_eq!(pool.selected.data_disk_gib(), selected, "{platform}");
            }

            let settings = data_disk_selection(Some(5000));
            let plan = plan_compute(&stack, platform, Some(&settings)).expect("plan");
            assert_eq!(
                general_pool(&plan).errors,
                vec!["pool 'general' data disk must be between 0 and 4096 GiB; dataDiskGib 5000 is outside that range".to_string()],
                "{platform}"
            );
        }
    }

    #[test]
    fn a_fixed_or_absent_data_disk_refuses_another_size() {
        let fixed = stack_with_data_disk_pool(Some(100), None);
        let plan = plan_compute(&fixed, Platform::Aws, None).expect("plan");
        assert_eq!(
            general_pool(&plan).data_disk,
            Some(ComputePoolDataDiskPlan {
                min_gib: 100,
                max_gib: 100,
                default_gib: 100,
                selected_gib: 100,
            })
        );
        let plan = plan_compute(&fixed, Platform::Aws, Some(&data_disk_selection(Some(200))))
            .expect("plan");
        assert_eq!(
            general_pool(&plan).errors,
            vec!["pool 'general' has a fixed 100 GiB data disk; dataDiskGib 200 cannot be chosen".to_string()]
        );

        let absent = stack_with_data_disk_pool(None, None);
        let plan = plan_compute(&absent, Platform::Aws, Some(&data_disk_selection(Some(0))))
            .expect("plan");
        assert!(general_pool(&plan).errors.is_empty());
        assert_eq!(general_pool(&plan).data_disk, None);
        let plan = plan_compute(&absent, Platform::Aws, Some(&data_disk_selection(Some(50))))
            .expect("plan");
        assert_eq!(
            general_pool(&plan).errors,
            vec!["pool 'general' declares no data disk, so dataDiskGib 50 cannot be chosen".to_string()]
        );
    }

    #[test]
    fn materialized_machine_profile_is_not_counted_as_workload_demand() {
        let mut stack = stack_with_container();
        let selected = instance_catalog::find_instance_type(Platform::Aws, "m7g.xlarge")
            .expect("machine should exist in catalog");
        let cluster = ComputeCluster::new("compute".to_string())
            .capacity_group(CapacityGroup {
                group_id: "general".to_string(),
                instance_type: Some(selected.name.to_string()),
                profile: Some(selected.to_machine_profile()),
                min_size: 1,
                max_size: 1,
                scale_policy: None,
                nested_virtualization: None,
                data_disk_gib_range: None,
            })
            .build();
        stack.resources.insert(
            "compute".to_string(),
            ResourceEntry {
                config: Resource::new(cluster),
                lifecycle: ResourceLifecycle::Frozen,
                dependencies: Vec::new(),
                display_name: None,
                description: None,
                remote_access: false,
                enabled_when: None,
            },
        );

        let plan = plan_compute(&stack, Platform::Aws, None).expect("plan should build");
        let pool = plan.pools.first().expect("general pool should exist");

        assert_eq!(pool.requirements.cpu, "2");
        assert_eq!(pool.requirements.memory_bytes, 4 * 1024 * 1024 * 1024);
        assert_eq!(pool.recommended.machine(), Some("m7g.xlarge"));
        assert!(pool
            .machines
            .iter()
            .any(|option| option.machine == "m7g.xlarge"));
    }

    #[test]
    fn nested_x86_fixed_range_pool_preserves_bounds_and_rejects_graviton() {
        let daemon = Daemon::new("vm-runtime-loader".to_string())
            .code(DaemonCode::Image {
                image: "example.com/vm-runtime:latest".to_string(),
            })
            .cluster("vm-runtime".to_string())
            .cpu(ResourceSpec {
                min: "2".to_string(),
                desired: "2".to_string(),
            })
            .memory(ResourceSpec {
                min: "4Gi".to_string(),
                desired: "4Gi".to_string(),
            })
            .permissions("loader".to_string())
            .build();
        let cluster = ComputeCluster::new("vm-runtime".to_string())
            .capacity_group(CapacityGroup {
                group_id: "general".to_string(),
                instance_type: None,
                profile: Some(MachineProfile {
                    cpu: "4".to_string(),
                    memory_bytes: 16 * 1024 * 1024 * 1024,
                    ephemeral_storage_bytes: 20 * 1024 * 1024 * 1024,
                    architecture: Some(Architecture::X86_64),
                    gpu: None,
                    data_disk: None,
                }),
                min_size: 2,
                max_size: 2,
                scale_policy: Some(CapacityGroupScalePolicy::Fixed {
                    machines: ComputeChoiceRange {
                        min: 1,
                        max: 5,
                        default: 2,
                    },
                }),
                nested_virtualization: Some(true),
                data_disk_gib_range: None,
            })
            .build();
        let stack = Stack {
            id: "vm-runtime".to_string(),
            resources: [
                (
                    "vm-runtime-loader".to_string(),
                    ResourceEntry {
                        config: Resource::new(daemon),
                        lifecycle: ResourceLifecycle::Live,
                        dependencies: Vec::new(),
                        display_name: None,
                        description: None,
                        remote_access: false,
                        enabled_when: None,
                    },
                ),
                (
                    "vm-runtime".to_string(),
                    ResourceEntry {
                        config: Resource::new(cluster),
                        lifecycle: ResourceLifecycle::Frozen,
                        dependencies: Vec::new(),
                        display_name: None,
                        description: None,
                        remote_access: false,
                        enabled_when: None,
                    },
                ),
            ]
            .into_iter()
            .collect(),
            permissions: crate::permissions::PermissionsConfig::default(),
            supported_platforms: None,
            inputs: vec![],
            dynamic_container_repositories: Vec::new(),
            dynamic_container_image_resources: Vec::new(),
            operations: None,
        };

        let plan = plan_compute(&stack, Platform::Aws, None).expect("plan should build");
        let pool = plan.pools.first().expect("general pool should exist");
        assert_eq!(pool.recommended.machine(), Some("m8i.2xlarge"));
        assert_eq!(pool.recommended.min_size(), 2);
        assert_eq!(pool.recommended.max_size(), 2);
        assert_eq!(
            pool.scale,
            CapacityGroupScalePolicy::Fixed {
                machines: ComputeChoiceRange {
                    min: 1,
                    max: 5,
                    default: 2,
                },
            }
        );
        assert!(!pool
            .machines
            .iter()
            .any(|option| option.machine == "m7g.2xlarge"));

        let invalid_settings = ComputeSettings {
            containers: Default::default(),
            pools: [(
                "general".to_string(),
                ComputePoolSelection::Fixed {
                    machines: 2,
                    machine: Some("m7g.2xlarge".to_string()),
                    failure_domains: None,
                    data_disk_gib: None,
                },
            )]
            .into_iter()
            .collect(),
        };
        let invalid_plan = plan_compute(&stack, Platform::Aws, Some(&invalid_settings))
            .expect("plan should build");
        assert!(!invalid_plan.pools[0].errors.is_empty());
    }

    fn declared_pool_errors(
        scale: CapacityGroupScalePolicy,
        selection: ComputePoolSelection,
    ) -> Vec<String> {
        let mut stack = stack_with_container();
        let cluster = ComputeCluster::new("compute".to_string())
            .capacity_group(CapacityGroup {
                group_id: "general".to_string(),
                instance_type: None,
                profile: None,
                min_size: scale.default_min_size(),
                max_size: scale.default_max_size(),
                scale_policy: Some(scale),
                nested_virtualization: None,
                data_disk_gib_range: None,
            })
            .build();
        stack.resources.insert(
            "compute".to_string(),
            ResourceEntry {
                config: Resource::new(cluster),
                lifecycle: ResourceLifecycle::Frozen,
                dependencies: Vec::new(),
                display_name: None,
                description: None,
                remote_access: false,
                enabled_when: None,
            },
        );
        let settings = ComputeSettings {
            containers: Default::default(),
            pools: [("general".to_string(), selection)].into_iter().collect(),
        };
        plan_compute(&stack, Platform::Aws, Some(&settings))
            .expect("plan should build")
            .pools
            .remove(0)
            .errors
    }

    fn range(min: u32, max: u32, default: u32) -> ComputeChoiceRange {
        ComputeChoiceRange { min, max, default }
    }

    #[test]
    fn generated_pool_offers_the_recommendation_as_a_default_not_a_bound() {
        let mut stack = stack_with_container();
        let api = stack
            .resources
            .get_mut("api")
            .and_then(|entry| entry.config.downcast_mut::<Container>())
            .expect("test stack should contain a container");
        api.autoscaling = Some(crate::ContainerAutoscaling {
            min: 1,
            desired: 1,
            max: 3,
            target_cpu_percent: Some(70.0),
            target_memory_percent: None,
            target_http_in_flight_per_replica: None,
            max_http_p95_latency_ms: None,
        });

        let plan = plan_compute(&stack, Platform::Aws, None).expect("plan should build");
        let pool = &plan.pools[0];
        assert_eq!(
            pool.scale,
            CapacityGroupScalePolicy::Autoscale {
                min: range(1, 100, 1),
                max: range(1, 100, 3),
            }
        );
        assert_eq!(
            (pool.recommended.min_size(), pool.recommended.max_size()),
            (1, 3)
        );

        // Workloads that never scale get a fixed recommendation, but either mode stays
        // allowed: replica counts change between releases while the saved choice stays.
        let plan =
            plan_compute(&stack_with_container(), Platform::Aws, None).expect("plan should build");
        assert_eq!(
            plan.pools[0].scale,
            CapacityGroupScalePolicy::Autoscale {
                min: range(1, 100, 1),
                max: range(1, 100, 1),
            }
        );
        assert!(matches!(
            plan.pools[0].recommended,
            ComputePoolSelection::Fixed { machines: 1, .. }
        ));
        assert!(plan.pools[0].errors.is_empty());
    }

    #[test]
    fn generated_pool_rejects_a_fleet_too_small_for_desired_replicas() {
        let mut stack = stack_with_container();
        let api = stack
            .resources
            .get_mut("api")
            .and_then(|entry| entry.config.downcast_mut::<Container>())
            .expect("test stack should contain a container");
        // 3 x 2 vCPU / 4 GiB requested at the desired count.
        api.replicas = Some(3);
        let selection = |machines| ComputeSettings {
            containers: Default::default(),
            pools: [(
                "general".to_string(),
                ComputePoolSelection::Fixed {
                    machines,
                    machine: Some("m7g.xlarge".to_string()),
                    failure_domains: None,
                    data_disk_gib: None,
                },
            )]
            .into_iter()
            .collect(),
        };

        // m7g.xlarge: 3.5 vCPU available after the host reserve.
        let errors = plan_compute(&stack, Platform::Aws, Some(&selection(1)))
            .expect("plan should build")
            .pools
            .remove(0)
            .errors;
        assert_eq!(errors.len(), 1, "{errors:?}");
        assert!(
            errors[0].starts_with(
                "Pool 'general' is too small for its workloads: 1 x m7g.xlarge has 3.50 vCPU"
            ),
            "{errors:?}"
        );
        assert!(errors[0].contains("request 6.00 vCPU"), "{errors:?}");

        let errors = plan_compute(&stack, Platform::Aws, Some(&selection(2)))
            .expect("plan should build")
            .pools
            .remove(0)
            .errors;
        assert!(errors.is_empty(), "{errors:?}");
    }

    #[test]
    fn fixed_and_equal_autoscale_bounds_are_interchangeable_for_declared_pools() {
        let fixed = |machines| ComputePoolSelection::Fixed {
            machines,
            machine: Some("m7g.xlarge".to_string()),
            failure_domains: None,
            data_disk_gib: None,
        };
        let autoscale = |min, max| ComputePoolSelection::Autoscale {
            min,
            max,
            machine: Some("m7g.xlarge".to_string()),
            failure_domains: None,
            data_disk_gib: None,
        };
        let declared_fixed = CapacityGroupScalePolicy::Fixed {
            machines: range(2, 4, 2),
        };
        let declared_autoscale = CapacityGroupScalePolicy::Autoscale {
            min: range(1, 3, 1),
            max: range(2, 6, 4),
        };

        assert!(declared_pool_errors(declared_fixed.clone(), autoscale(3, 3)).is_empty());
        assert_eq!(
            declared_pool_errors(declared_fixed.clone(), autoscale(2, 4)),
            vec!["Pool 'general' must use fixed mode with 2-4 machines; autoscale 2-4 is not allowed"]
        );
        assert_eq!(
            declared_pool_errors(declared_fixed, autoscale(5, 5)),
            vec!["Pool 'general' machine count 5 is outside the allowed range 2-4"]
        );

        // Fixed N must fit both the declared minimum and maximum ranges.
        assert!(declared_pool_errors(declared_autoscale.clone(), fixed(3)).is_empty());
        assert_eq!(
            declared_pool_errors(declared_autoscale.clone(), fixed(1)),
            vec!["Pool 'general' fixed machine count 1 is outside the allowed range 2-3"]
        );
        assert_eq!(
            declared_pool_errors(
                CapacityGroupScalePolicy::Autoscale {
                    min: range(1, 1, 1),
                    max: range(3, 5, 3),
                },
                fixed(1),
            ),
            vec!["Pool 'general' must use autoscale mode (minimum 1-1, maximum 3-5)"]
        );
        assert_eq!(
            declared_pool_errors(declared_autoscale, autoscale(1, 7)),
            vec!["Pool 'general' autoscale maximum 7 is outside the allowed range 2-6"]
        );
    }

    #[test]
    fn local_plan_has_no_provider_machine_choices() {
        let stack = stack_with_container();

        let plan = plan_compute(&stack, Platform::Local, None).expect("plan should build");

        let pool = plan.pools.first().expect("general pool should exist");
        assert_eq!(pool.selected.machine(), None);
        assert!(pool.machines.is_empty());
        assert!(pool.errors.is_empty());
    }
}