zlayer-agent 0.14.1

//! `AutoscaleController` - Connects autoscaling decisions to container scaling
//!
//! This module provides an `AutoscaleController` that bridges the scheduler's
//! autoscaling logic with the agent's `ServiceManager` to automatically scale
//! services based on resource utilization.
//!
//! It drives three orthogonal control loops over the same metrics tick:
//!
//! 1. **Horizontal** (replica count) — the original adaptive autoscaler that
//!    grows/shrinks replicas to hit CPU / memory / RPS targets.
//! 2. **Scale-to-zero** (Phase 2) — when an adaptive service declares
//!    `idle_window` and `min == 0`, the controller reaps every replica after
//!    the service has been idle (no meaningful CPU / RPS) for the window. The
//!    proxy activator wakes it again by calling [`AutoscaleController::mark_active`]
//!    on the next inbound request.
//! 3. **Vertical** (right-sizing, Phase 3) — when an adaptive service declares
//!    a `vertical` block, a [`VpaEngine`] observes per-replica usage and emits
//!    CPU-millis / memory-MiB recommendations. In `Recommend` mode they are
//!    logged; in `Auto` mode they are applied via
//!    [`Runtime::update_container_resources`], with a rolling restart fallback
//!    when the runtime cannot live-update a running container's cgroup.
//!
//! # Architecture
//!
//! ```text
//! ┌────────────────────────────────────────────────────────────────────┐
//! │                     AutoscaleController                            │
//! │  ┌─────────────────┐  ┌────────────┐  ┌──────────────────┐       │
//! │  │ CgroupsMetrics  │  │ Autoscaler │  │ ServiceManager   │       │
//! │  │    Source       │──│  + VpaEngine│──│  (scaling)       │       │
//! │  └─────────────────┘  └────────────┘  └──────────────────┘       │
//! └────────────────────────────────────────────────────────────────────┘
//! ```
//!
//! # Example
//!
//! ```ignore
//! use zlayer_agent::autoscale_controller::AutoscaleController;
//! use zlayer_agent::{ServiceManager, RuntimeConfig, create_runtime};
//! use std::sync::Arc;
//! use std::time::Duration;
//!
//! // Create runtime and service manager
//! let runtime = create_runtime(RuntimeConfig::Mock).await?;
//! let manager = Arc::new(ServiceManager::new(runtime.clone()));
//!
//! // Create autoscale controller
//! let controller = AutoscaleController::new(
//!     manager.clone(),
//!     runtime.clone(),
//!     Duration::from_secs(10),
//! );
//!
//! // Register services with adaptive scaling
//! controller.register_service("api", &scale_spec, 2).await;
//!
//! // Run the autoscaling loop (in background)
//! let handle = tokio::spawn(async move {
//!     controller.run_loop().await
//! });
//!
//! // Later, shutdown
//! controller.shutdown();
//! ```

use crate::error::{AgentError, Result};
use crate::metrics_providers::{LockedServiceManagerContainerProvider, RuntimeStatsProvider};
use crate::runtime::{ContainerId, ContainerResourceUpdate, Runtime};
use crate::service::ServiceManager;
use std::collections::HashMap;
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::sync::RwLock;
use tracing::{debug, error, info, warn};
use zlayer_scheduler::metrics::{
    CgroupsMetricsSource, ContainerStatsProvider, MetricsCollector, MetricsContainerId,
    MetricsSource, RawContainerStats,
};
use zlayer_scheduler::Autoscaler;
use zlayer_spec::{ScaleSpec, ServiceSpec, VerticalMode, VerticalScaleSpec};

/// Default autoscaling evaluation interval
pub const DEFAULT_AUTOSCALE_INTERVAL: Duration = Duration::from_secs(10);

/// CPU usage (microseconds-per-second-equivalent) below which a replica is
/// considered idle for scale-to-zero accounting. The cgroups stats provider
/// reports cumulative `cpu_usage_usec`; the [`VpaEngine`] turns successive
/// samples into a per-second rate, and any service whose busiest replica
/// stays under this rate is treated as "no meaningful work".
///
/// `5_000` µs/s ≈ 0.5% of one core — low enough to ignore idle bookkeeping
/// (health probes, GC ticks) but high enough that a real request bumps it.
const IDLE_CPU_RATE_USEC_PER_SEC: f64 = 5_000.0;

/// Fraction by which a fresh vertical recommendation must differ from the
/// last-applied value before it is re-applied. Prevents thrashing the cgroup
/// hierarchy (and, in the rolling-restart fallback, the containers themselves)
/// on sub-percent jitter. 0.10 == "only act on a ≥10% change".
const VERTICAL_DEADBAND: f64 = 0.10;

/// A vertical right-sizing recommendation for a single container: the target
/// CPU allotment in millicores and the target memory in MiB.
///
/// Mirrors the shape the agent applies through
/// [`Runtime::update_container_resources`]: `cpu_millis` becomes a CFS
/// quota/period pair and `memory_mib` a byte limit.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct VpaRecommendation {
    /// Recommended CPU allotment in millicores (1000 == one full core).
    pub cpu_millis: u32,
    /// Recommended memory limit in MiB.
    pub memory_mib: u32,
}

/// Per-container vertical-pod-autoscaler engine.
///
/// Keeps a small ring buffer of recent usage samples per container, derives a
/// CPU rate (millicores) from successive cumulative-CPU readings, and emits a
/// percentile-based recommendation clamped to the service's
/// [`VerticalScaleSpec`] bounds. This is the agent-local engine used by
/// [`AutoscaleController`]; it deliberately lives here (rather than in
/// `zlayer-scheduler`) so the whole vertical-apply path is self-contained in
/// the controller.
#[derive(Debug, Default)]
pub struct VpaEngine {
    /// Per-container usage history, keyed by the container's display id.
    history: HashMap<String, ContainerUsageHistory>,
}

/// Rolling usage history for a single container. Bounded to the most recent
/// [`Self::CAPACITY`] samples so the percentile computation has a stable,
/// memory-bounded window.
#[derive(Debug, Default)]
struct ContainerUsageHistory {
    /// Recent CPU rates in millicores, oldest first.
    cpu_millis: std::collections::VecDeque<f64>,
    /// Recent memory readings in MiB, oldest first.
    memory_mib: std::collections::VecDeque<f64>,
    /// Last cumulative CPU reading (µs) + the wallclock instant it was taken,
    /// used to derive a rate from the next sample.
    last_cpu: Option<(u64, Instant)>,
}

impl ContainerUsageHistory {
    /// Maximum number of samples retained per container.
    const CAPACITY: usize = 32;

    fn push_cpu(&mut self, millis: f64) {
        if self.cpu_millis.len() == Self::CAPACITY {
            self.cpu_millis.pop_front();
        }
        self.cpu_millis.push_back(millis);
    }

    fn push_memory(&mut self, mib: f64) {
        if self.memory_mib.len() == Self::CAPACITY {
            self.memory_mib.pop_front();
        }
        self.memory_mib.push_back(mib);
    }

    /// Percentile (0-100) over a sample window, nearest-rank. Returns `None`
    /// when the window is empty.
    fn percentile(samples: &std::collections::VecDeque<f64>, pct: u8) -> Option<f64> {
        if samples.is_empty() {
            return None;
        }
        let mut sorted: Vec<f64> = samples.iter().copied().collect();
        sorted.sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal));
        let pct = f64::from(pct.min(100)) / 100.0;
        // Nearest-rank: index = ceil(pct * n) - 1, clamped into range.
        #[allow(
            clippy::cast_possible_truncation,
            clippy::cast_sign_loss,
            clippy::cast_precision_loss
        )]
        let idx = ((pct * sorted.len() as f64).ceil() as usize)
            .saturating_sub(1)
            .min(sorted.len() - 1);
        Some(sorted[idx])
    }
}

impl VpaEngine {
    /// Construct an empty engine.
    #[must_use]
    pub fn new() -> Self {
        Self::default()
    }

    /// Record one raw stats sample for `container` and return the busiest-replica
    /// CPU rate (millicores) this sample implies, so callers can drive
    /// scale-to-zero idle accounting off the same observation.
    ///
    /// CPU rate is derived from the delta between this sample's cumulative
    /// `cpu_usage_usec` and the previous one over the elapsed wallclock time:
    /// `millis = (Δusec / Δsec) / 1000`. The first sample for a container has no
    /// predecessor, so it contributes a memory reading but a `0.0` CPU rate.
    pub fn observe(&mut self, container: &str, stats: &RawContainerStats) -> f64 {
        let now = Instant::now();
        let hist = self.history.entry(container.to_string()).or_default();

        let cpu_millis = if let Some((prev_usec, prev_at)) = hist.last_cpu {
            let elapsed = now.duration_since(prev_at).as_secs_f64();
            let delta_usec = stats.cpu_usage_usec.saturating_sub(prev_usec);
            if elapsed > 0.0 {
                // (µs busy / s) / 1000 µs-per-ms-of-core == millicores.
                #[allow(clippy::cast_precision_loss)]
                let rate = delta_usec as f64 / elapsed / 1000.0;
                hist.push_cpu(rate);
                rate
            } else {
                0.0
            }
        } else {
            0.0
        };
        hist.last_cpu = Some((stats.cpu_usage_usec, now));

        #[allow(clippy::cast_precision_loss)]
        let mem_mib = stats.memory_bytes as f64 / (1024.0 * 1024.0);
        hist.push_memory(mem_mib);

        cpu_millis
    }

    /// Compute a recommendation for `container` from its observed history,
    /// clamped to `spec`'s bounds. Returns `None` until enough samples exist to
    /// produce a stable percentile (at least one CPU rate observation).
    #[must_use]
    pub fn recommend(
        &self,
        container: &str,
        spec: &VerticalScaleSpec,
    ) -> Option<VpaRecommendation> {
        let hist = self.history.get(container)?;
        let cpu_pct = ContainerUsageHistory::percentile(&hist.cpu_millis, spec.percentile)?;
        let mem_pct = ContainerUsageHistory::percentile(&hist.memory_mib, spec.percentile)?;

        // Round up to whole units and clamp to the configured bounds.
        #[allow(
            clippy::cast_possible_truncation,
            clippy::cast_sign_loss,
            clippy::cast_precision_loss
        )]
        let cpu_millis = {
            let mut v = cpu_pct.ceil().max(0.0) as u32;
            if let Some(min) = spec.min_cpu_millis {
                v = v.max(min);
            }
            if let Some(max) = spec.max_cpu_millis {
                v = v.min(max);
            }
            v.max(1)
        };
        #[allow(
            clippy::cast_possible_truncation,
            clippy::cast_sign_loss,
            clippy::cast_precision_loss
        )]
        let memory_mib = {
            let mut v = mem_pct.ceil().max(0.0) as u32;
            if let Some(min) = spec.min_memory_mib {
                v = v.max(min);
            }
            if let Some(max) = spec.max_memory_mib {
                v = v.min(max);
            }
            v.max(1)
        };

        Some(VpaRecommendation {
            cpu_millis,
            memory_mib,
        })
    }

    /// Drop all history for a container (e.g. after it is removed by a
    /// scale-down or rolling restart) so a recreated replica starts fresh.
    pub fn forget(&mut self, container: &str) {
        self.history.remove(container);
    }
}

/// True when a fresh recommendation differs from the last-applied one by more
/// than [`VERTICAL_DEADBAND`] on either axis. A `None` previous value always
/// passes (first application).
fn outside_deadband(prev: Option<VpaRecommendation>, next: VpaRecommendation) -> bool {
    let Some(prev) = prev else { return true };
    let exceeds = |old: u32, new: u32| {
        if old == 0 {
            return new != 0;
        }
        let delta = (f64::from(new) - f64::from(old)).abs();
        delta / f64::from(old) > VERTICAL_DEADBAND
    };
    exceeds(prev.cpu_millis, next.cpu_millis) || exceeds(prev.memory_mib, next.memory_mib)
}

/// Build the [`ContainerResourceUpdate`] that applies `rec` to a container:
/// `cpu_millis` becomes a CFS quota over a fixed 100 ms period, and
/// `memory_mib` becomes a byte limit.
fn resource_update_for(rec: VpaRecommendation) -> ContainerResourceUpdate {
    ContainerResourceUpdate {
        cpu_period: Some(100_000),
        cpu_quota: Some(100_000 * i64::from(rec.cpu_millis) / 1000),
        memory: Some(i64::from(rec.memory_mib) * 1024 * 1024),
        ..Default::default()
    }
}

/// Controller that connects autoscaling decisions to actual container scaling
///
/// The `AutoscaleController` periodically collects metrics from running containers,
/// evaluates whether scaling is needed using the `Autoscaler`, and executes scaling
/// decisions through the `ServiceManager`. On the same tick it runs the
/// scale-to-zero idle reaper and the vertical right-sizing pass.
pub struct AutoscaleController {
    /// Service manager for executing scaling operations. Held as the daemon's
    /// post-`Arc::try_unwrap` `Arc<RwLock<ServiceManager>>`; each call site takes
    /// a short read guard.
    service_manager: Arc<RwLock<ServiceManager>>,
    /// Metrics collector with cgroups source
    metrics: Arc<MetricsCollector>,
    /// Autoscaler decision engine
    autoscaler: Arc<RwLock<Autoscaler>>,
    /// Service specs for scale configuration (`service_name` -> spec)
    service_specs: Arc<RwLock<HashMap<String, ScaleSpec>>>,
    /// Last scale times for cooldown tracking (`service_name` -> instant)
    last_scale_times: Arc<RwLock<HashMap<String, Instant>>>,
    /// Evaluation interval
    interval: Duration,
    /// Shutdown signal
    shutdown: Arc<tokio::sync::Notify>,
    // --- Phase 2 / Phase 3 state (kept at the struct end so the horizontal
    //     path above is untouched) ---
    /// Container runtime handle, retained so the vertical-apply path can call
    /// [`Runtime::update_container_resources`] and the rolling-restart fallback
    /// can recreate containers. Cloned before the same `Arc` is moved into the
    /// stats provider in [`AutoscaleController::new`].
    runtime: Arc<dyn Runtime + Send + Sync>,
    /// The stats provider wrapping the runtime, retained so the vertical pass
    /// can fetch per-replica [`RawContainerStats`] directly (the horizontal
    /// path goes through the aggregating [`MetricsCollector`]).
    stats_provider: Arc<RuntimeStatsProvider>,
    /// Last time each service showed meaningful activity, for scale-to-zero.
    last_active: Arc<RwLock<HashMap<String, Instant>>>,
    /// Idle window per service (`Some` enables scale-to-zero when `min == 0`).
    idle_window: Arc<RwLock<HashMap<String, Duration>>>,
    /// Minimum replicas per service (scale-to-zero only fires when `min == 0`).
    min_replicas: Arc<RwLock<HashMap<String, u32>>>,
    /// Vertical right-sizing spec per service (`Some` enables the VPA pass).
    vertical_specs: Arc<RwLock<HashMap<String, VerticalScaleSpec>>>,
    /// Full base [`ServiceSpec`] per service, supplied via
    /// [`AutoscaleController::set_service_template`]. Required for the
    /// rolling-restart fallback so a recreated replica keeps its endpoints,
    /// env, volumes, etc. while picking up the new resources.
    service_templates: Arc<RwLock<HashMap<String, ServiceSpec>>>,
    /// Vertical decision engine + last-applied recommendation per service, used
    /// to apply the deadband.
    vpa: Arc<RwLock<VpaState>>,
}

/// Internal vertical-autoscaler state: the engine plus the last recommendation
/// actually applied to each service (for deadband comparison).
#[derive(Default)]
struct VpaState {
    engine: VpaEngine,
    last_applied: HashMap<String, VpaRecommendation>,
}

impl AutoscaleController {
    /// Create a new autoscale controller
    ///
    /// # Arguments
    /// * `service_manager` - The service manager used to execute scaling operations
    /// * `runtime` - The container runtime for collecting metrics and applying
    ///   vertical resource updates
    /// * `interval` - How often to evaluate scaling decisions
    ///
    /// The `runtime` handle is retained on the controller (for the vertical
    /// apply / rolling-restart paths) *and* moved into the metrics stats
    /// provider, so it is cloned once here.
    ///
    /// # Example
    ///
    /// ```ignore
    /// let controller = AutoscaleController::new(
    ///     service_manager,
    ///     runtime,
    ///     Duration::from_secs(10),
    /// );
    /// ```
    #[must_use]
    pub fn new(
        service_manager: Arc<RwLock<ServiceManager>>,
        runtime: Arc<dyn Runtime + Send + Sync>,
        interval: Duration,
    ) -> Self {
        // Create metrics collector with cgroups source
        let mut metrics = MetricsCollector::new();

        // Create the stats provider wrapping the runtime. Clone the runtime
        // first so the controller keeps its own handle for vertical apply.
        let runtime_for_controller = runtime.clone();
        let stats_provider = Arc::new(RuntimeStatsProvider::new(runtime));

        // Create the service container provider wrapping the locked service
        // manager.
        let service_provider = Arc::new(LockedServiceManagerContainerProvider::new(
            service_manager.clone(),
        ));

        // Create cgroups metrics source
        let source: Arc<dyn MetricsSource> = Arc::new(CgroupsMetricsSource::new(
            service_provider,
            stats_provider.clone(),
        ));
        metrics.add_source(source);

        Self {
            service_manager,
            metrics: Arc::new(metrics),
            autoscaler: Arc::new(RwLock::new(Autoscaler::new())),
            service_specs: Arc::new(RwLock::new(HashMap::new())),
            last_scale_times: Arc::new(RwLock::new(HashMap::new())),
            interval,
            shutdown: Arc::new(tokio::sync::Notify::new()),
            runtime: runtime_for_controller,
            stats_provider,
            last_active: Arc::new(RwLock::new(HashMap::new())),
            idle_window: Arc::new(RwLock::new(HashMap::new())),
            min_replicas: Arc::new(RwLock::new(HashMap::new())),
            vertical_specs: Arc::new(RwLock::new(HashMap::new())),
            service_templates: Arc::new(RwLock::new(HashMap::new())),
            vpa: Arc::new(RwLock::new(VpaState::default())),
        }
    }

    /// Create with a custom metrics collector (useful for testing).
    ///
    /// Requires a `runtime` handle so the vertical-apply path is still wired in
    /// tests; pass a mock runtime when only the horizontal/scale-to-zero paths
    /// are under test.
    #[must_use]
    pub fn with_custom_metrics(
        service_manager: Arc<RwLock<ServiceManager>>,
        runtime: Arc<dyn Runtime + Send + Sync>,
        metrics: MetricsCollector,
        interval: Duration,
    ) -> Self {
        let runtime_for_controller = runtime.clone();
        let stats_provider = Arc::new(RuntimeStatsProvider::new(runtime));
        Self {
            service_manager,
            metrics: Arc::new(metrics),
            autoscaler: Arc::new(RwLock::new(Autoscaler::new())),
            service_specs: Arc::new(RwLock::new(HashMap::new())),
            last_scale_times: Arc::new(RwLock::new(HashMap::new())),
            interval,
            shutdown: Arc::new(tokio::sync::Notify::new()),
            runtime: runtime_for_controller,
            stats_provider,
            last_active: Arc::new(RwLock::new(HashMap::new())),
            idle_window: Arc::new(RwLock::new(HashMap::new())),
            min_replicas: Arc::new(RwLock::new(HashMap::new())),
            vertical_specs: Arc::new(RwLock::new(HashMap::new())),
            service_templates: Arc::new(RwLock::new(HashMap::new())),
            vpa: Arc::new(RwLock::new(VpaState::default())),
        }
    }

    /// Push an additional [`MetricsSource`] into the controller's collector.
    ///
    /// The daemon uses this to feed real requests-per-second from the L7 proxy
    /// (wrapped in a [`zlayer_scheduler::metrics::ProxyRpsMetricsSource`]) into
    /// the same [`MetricsCollector`] that drives the horizontal pass, so
    /// [`zlayer_scheduler::metrics::AggregatedMetrics::total_rps`] is populated
    /// and RPS targets / triggers fire on live traffic.
    ///
    /// This is a builder method that must be called **before** the controller is
    /// shared (the collector is still uniquely owned at construction time, so
    /// the in-place mutation always succeeds). If the collector has already been
    /// cloned, the source is dropped with a warning rather than panicking.
    #[must_use]
    pub fn with_extra_metrics_source(mut self, source: Arc<dyn MetricsSource>) -> Self {
        if let Some(collector) = Arc::get_mut(&mut self.metrics) {
            collector.add_source(source);
        } else {
            warn!(
                "with_extra_metrics_source called after the metrics collector was shared; \
                 source ignored"
            );
        }
        self
    }

    /// Register a service for autoscaling
    ///
    /// Only services with `ScaleSpec::Adaptive` will be evaluated for autoscaling.
    /// Services with `Fixed` or `Manual` scaling are ignored by the autoscaler loop.
    ///
    /// Beyond the horizontal registration, this captures the Phase 2 / Phase 3
    /// configuration carried on the adaptive spec: `idle_window` + `min` (for
    /// scale-to-zero) and `vertical` (for right-sizing). The service is seeded
    /// as freshly-active so a just-registered service is not immediately reaped.
    ///
    /// # Arguments
    /// * `name` - Service name
    /// * `spec` - The service's scale specification
    /// * `initial_replicas` - Current number of replicas
    pub async fn register_service(&self, name: &str, spec: &ScaleSpec, initial_replicas: u32) {
        // Only register adaptive services
        let ScaleSpec::Adaptive {
            min,
            idle_window,
            vertical,
            ..
        } = spec
        else {
            debug!(
                service = name,
                "Skipping registration for non-adaptive service"
            );
            return;
        };

        // Register with autoscaler
        {
            let mut autoscaler = self.autoscaler.write().await;
            autoscaler.register_service(name, spec.clone(), initial_replicas);
        }

        // Store spec for reference
        {
            let mut specs = self.service_specs.write().await;
            specs.insert(name.to_string(), spec.clone());
        }

        // Phase 2: scale-to-zero bookkeeping.
        {
            let mut mins = self.min_replicas.write().await;
            mins.insert(name.to_string(), *min);
        }
        if let Some(window) = idle_window {
            self.idle_window
                .write()
                .await
                .insert(name.to_string(), *window);
        } else {
            self.idle_window.write().await.remove(name);
        }
        // Seed activity so a just-registered service has a full idle window
        // before it can be reaped.
        self.last_active
            .write()
            .await
            .insert(name.to_string(), Instant::now());

        // Phase 3: vertical right-sizing.
        if let Some(v) = vertical {
            if matches!(v.mode, VerticalMode::Recommend | VerticalMode::Auto) {
                self.vertical_specs
                    .write()
                    .await
                    .insert(name.to_string(), v.clone());
            } else {
                self.vertical_specs.write().await.remove(name);
            }
        } else {
            self.vertical_specs.write().await.remove(name);
        }

        info!(
            service = name,
            initial_replicas,
            idle_window_secs = idle_window.as_ref().map(Duration::as_secs),
            min = *min,
            vertical = vertical.is_some(),
            "Registered service for autoscaling"
        );
    }

    /// Supply (or refresh) the full base [`ServiceSpec`] for a service.
    ///
    /// Optional, but required for the vertical-apply **rolling restart**
    /// fallback (`update_container_resources` → `Unsupported`): recreating a
    /// replica needs its endpoints / env / volumes, which the [`ScaleSpec`]
    /// alone does not carry. The daemon wires this alongside
    /// [`AutoscaleController::register_service`]. Without a template, the
    /// fallback degrades to a full `scale 0 → scale n` bounce.
    pub async fn set_service_template(&self, name: &str, spec: ServiceSpec) {
        self.service_templates
            .write()
            .await
            .insert(name.to_string(), spec);
    }

    /// Mark a service as active *now*, resetting its scale-to-zero idle clock.
    ///
    /// Called by the proxy activator when an inbound request wakes (or keeps
    /// awake) a service so the idle reaper does not tear it down while it is
    /// actively serving traffic. Safe to call for services that are not
    /// registered for scale-to-zero — it simply records a timestamp that the
    /// idle pass ignores.
    pub fn mark_active(&self, service: &str) {
        // Synchronous entry point for callers outside an async context: take
        // the lock with `blocking_write` only when we must, but the common
        // path (inside the proxy's async task) should prefer the async helper.
        let last_active = self.last_active.clone();
        let service = service.to_string();
        if let Ok(mut guard) = last_active.try_write() {
            guard.insert(service, Instant::now());
            return;
        }
        // Lock contended: fall back to a detached task so we never block the
        // caller. The activation timestamp is still recorded promptly.
        tokio::spawn(async move {
            last_active.write().await.insert(service, Instant::now());
        });
    }

    /// Async variant of [`AutoscaleController::mark_active`] for callers already
    /// inside an async context (avoids the `try_write`/spawn dance).
    pub async fn mark_active_async(&self, service: &str) {
        self.last_active
            .write()
            .await
            .insert(service.to_string(), Instant::now());
    }

    /// Unregister a service from autoscaling
    pub async fn unregister_service(&self, name: &str) {
        {
            let mut autoscaler = self.autoscaler.write().await;
            autoscaler.unregister_service(name);
        }

        self.service_specs.write().await.remove(name);
        self.last_scale_times.write().await.remove(name);
        self.last_active.write().await.remove(name);
        self.idle_window.write().await.remove(name);
        self.min_replicas.write().await.remove(name);
        self.vertical_specs.write().await.remove(name);
        self.service_templates.write().await.remove(name);
        self.vpa.write().await.last_applied.remove(name);

        info!(service = name, "Unregistered service from autoscaling");
    }

    /// Check if a service is registered for autoscaling
    pub async fn is_registered(&self, name: &str) -> bool {
        let specs = self.service_specs.read().await;
        specs.contains_key(name)
    }

    /// Check if a service is in cooldown period
    ///
    /// Returns true if the service was scaled recently and is still in cooldown.
    async fn should_scale(&self, service_name: &str) -> bool {
        // Get the cooldown duration from the spec
        let cooldown = {
            let specs = self.service_specs.read().await;
            match specs.get(service_name) {
                Some(ScaleSpec::Adaptive { cooldown, .. }) => {
                    cooldown.unwrap_or(zlayer_scheduler::DEFAULT_COOLDOWN)
                }
                _ => return false, // Not adaptive, shouldn't scale
            }
        };

        // Check if we're past the cooldown period
        let last_scale_times = self.last_scale_times.read().await;
        if let Some(last_time) = last_scale_times.get(service_name) {
            if last_time.elapsed() < cooldown {
                let remaining = cooldown
                    .checked_sub(last_time.elapsed())
                    .unwrap_or_default();
                debug!(
                    service = service_name,
                    remaining_secs = remaining.as_secs(),
                    "Service in cooldown"
                );
                return false;
            }
        }

        true
    }

    /// Record that a scale action occurred
    async fn record_scale_action(&self, service_name: &str) {
        let mut times = self.last_scale_times.write().await;
        times.insert(service_name.to_string(), Instant::now());
    }

    /// Run the autoscaling loop
    ///
    /// This method should be spawned as a background task. It will continuously
    /// evaluate scaling decisions at the configured interval until shutdown is
    /// signaled.
    ///
    /// # Returns
    /// Returns `Ok(())` when shutdown is signaled, or an error if something
    /// goes wrong.
    ///
    /// # Example
    ///
    /// ```ignore
    /// let controller = Arc::new(AutoscaleController::new(...));
    /// let controller_clone = controller.clone();
    ///
    /// // Spawn the autoscale loop
    /// let handle = tokio::spawn(async move {
    ///     controller_clone.run_loop().await
    /// });
    ///
    /// // Later, shutdown
    /// controller.shutdown();
    /// handle.await.unwrap();
    /// ```
    /// # Errors
    /// Returns an error if the autoscale loop encounters an unrecoverable error.
    #[allow(clippy::cast_possible_truncation)]
    pub async fn run_loop(&self) -> Result<()> {
        let mut ticker = tokio::time::interval(self.interval);

        info!(
            interval_ms = self.interval.as_millis() as u64,
            "Starting autoscale controller loop"
        );

        loop {
            tokio::select! {
                _ = ticker.tick() => {
                    // Boxed: the evaluation future is large; keep it off the
                    // select!'s stack future (clippy::large_future).
                    Box::pin(self.evaluate_all_services()).await;
                }
                () = self.shutdown.notified() => {
                    info!("Autoscale controller shutting down");
                    break;
                }
            }
        }

        Ok(())
    }

    /// Self-discover services from the live [`ServiceManager`] and (re)register
    /// any adaptive service whose [`ScaleSpec`] is new or has changed since the
    /// last tick. Non-adaptive services are ignored by `register_service`.
    ///
    /// This is what makes the controller scale *real* services: the daemon never
    /// has to imperatively call `register_service` on deploy — every tick the
    /// controller reconciles its registration set against the manager's current
    /// view. A service whose adaptive spec is unchanged is left alone (so its
    /// cooldown / idle clocks are preserved); a removed service is unregistered.
    async fn discover_services(&self) {
        let live = self.service_manager.read().await.service_specs().await;

        // Register / refresh adaptive services.
        let mut seen_adaptive: Vec<String> = Vec::new();
        for (name, spec) in &live {
            if !matches!(spec.scale, ScaleSpec::Adaptive { .. }) {
                continue;
            }
            seen_adaptive.push(name.clone());

            // Only (re)register when the adaptive spec is new or changed, so we
            // don't reset cooldown/idle bookkeeping on every tick.
            let needs_register = {
                let specs = self.service_specs.read().await;
                specs.get(name) != Some(&spec.scale)
            };
            if needs_register {
                // Seed the initial replica count from the live service so the
                // autoscaler's internal state starts from reality.
                let initial = u32::try_from(
                    self.service_manager
                        .read()
                        .await
                        .service_replica_count(name)
                        .await
                        .unwrap_or(0),
                )
                .unwrap_or(0);
                self.register_service(name, &spec.scale, initial).await;
            }
            // Always refresh the full template so the rolling-restart fallback
            // and any vertical apply use the current spec (cheap clone).
            self.set_service_template(name, spec.clone()).await;
        }

        // Unregister services that vanished from the manager (or stopped being
        // adaptive) since the last tick.
        let registered: Vec<String> = {
            let specs = self.service_specs.read().await;
            specs.keys().cloned().collect()
        };
        for name in registered {
            if !seen_adaptive.contains(&name) {
                self.unregister_service(&name).await;
            }
        }
    }

    /// Evaluate and potentially scale all registered services
    async fn evaluate_all_services(&self) {
        // Reconcile the registration set against the live ServiceManager before
        // evaluating, so newly-deployed adaptive services start scaling without
        // any explicit registration call.
        self.discover_services().await;

        // Get list of registered services
        let service_names: Vec<String> = {
            let specs = self.service_specs.read().await;
            specs.keys().cloned().collect()
        };

        for service_name in service_names {
            // Vertical (right-sizing) pass first so a recommendation applied
            // this tick is reflected in the next horizontal observation.
            if let Err(e) = Box::pin(self.evaluate_vertical(&service_name)).await {
                warn!(
                    service = %service_name,
                    error = %e,
                    "Failed vertical (VPA) evaluation"
                );
            }

            // Scale-to-zero idle reaper. Runs before the horizontal pass so a
            // reaped service short-circuits the (now pointless) replica math.
            match self.evaluate_idle(&service_name).await {
                Ok(true) => continue, // Service was reaped to zero this tick.
                Ok(false) => {}
                Err(e) => warn!(
                    service = %service_name,
                    error = %e,
                    "Failed scale-to-zero evaluation"
                ),
            }

            if let Err(e) = self.evaluate_and_scale(&service_name).await {
                // Log but don't fail the entire loop
                warn!(
                    service = %service_name,
                    error = %e,
                    "Failed to evaluate/scale service"
                );
            }
        }
    }

    /// Scale-to-zero idle reaper (Phase 2).
    ///
    /// Bumps `last_active` whenever the service's busiest replica shows
    /// non-trivial CPU usage, then — for a service with `idle_window: Some(w)`
    /// and `min == 0` — tears every replica down once it has been idle longer
    /// than `w` (respecting the scaling cooldown). Returns `Ok(true)` when the
    /// service was reaped this tick so the caller can skip the horizontal pass.
    async fn evaluate_idle(&self, service_name: &str) -> Result<bool> {
        // Per-replica stats drive activity detection. The cgroups stats
        // provider reports cumulative CPU; the VPA engine turns that into a
        // per-second rate. Any replica over the idle threshold counts as
        // activity. (RPS-based activity flows through `mark_active` from the
        // proxy, so even a zero-CPU request keeps the service warm.)
        let containers = self
            .service_manager
            .read()
            .await
            .get_service_containers(service_name)
            .await;
        let mut busiest_cpu_millis = 0.0_f64;
        {
            let mut vpa = self.vpa.write().await;
            for id in &containers {
                let metrics_id = MetricsContainerId {
                    service: id.service.clone(),
                    replica: id.replica,
                };
                if let Ok(stats) = self.stats_provider.get_stats(&metrics_id).await {
                    let rate = vpa.engine.observe(&id.to_string(), &stats);
                    if rate > busiest_cpu_millis {
                        busiest_cpu_millis = rate;
                    }
                }
            }
        }

        if busiest_cpu_millis * 1000.0 >= IDLE_CPU_RATE_USEC_PER_SEC {
            self.last_active
                .write()
                .await
                .insert(service_name.to_string(), Instant::now());
        }

        // Scale-to-zero only applies when an idle window is configured *and*
        // the service's floor is zero.
        let window = {
            let windows = self.idle_window.read().await;
            match windows.get(service_name) {
                Some(w) => *w,
                None => return Ok(false),
            }
        };
        let min = self
            .min_replicas
            .read()
            .await
            .get(service_name)
            .copied()
            .unwrap_or(1);
        if min != 0 {
            return Ok(false);
        }

        // Nothing to reap if already at zero.
        let current = self
            .service_manager
            .read()
            .await
            .service_replica_count(service_name)
            .await
            .unwrap_or(0);
        if current == 0 {
            return Ok(false);
        }

        let idle_for = {
            let last_active = self.last_active.read().await;
            last_active
                .get(service_name)
                .map_or(Duration::ZERO, Instant::elapsed)
        };
        if idle_for <= window {
            return Ok(false);
        }

        // Respect the scaling cooldown so a flapping service doesn't get
        // reaped and re-woken in a tight loop.
        if !self.should_scale(service_name).await {
            return Ok(false);
        }

        info!(
            service = service_name,
            idle_secs = idle_for.as_secs(),
            window_secs = window.as_secs(),
            "Scaling service to zero (idle past window)"
        );
        self.service_manager
            .read()
            .await
            .scale_service(service_name, 0)
            .await?;
        self.record_scale_action(service_name).await;
        {
            let mut autoscaler = self.autoscaler.write().await;
            if let Err(e) = autoscaler.record_scale_action(service_name, 0) {
                warn!(
                    service = service_name,
                    error = %e,
                    "Failed to record scale-to-zero in autoscaler"
                );
            }
        }
        // Forget per-replica history so a re-woken service starts fresh.
        {
            let mut vpa = self.vpa.write().await;
            for id in &containers {
                vpa.engine.forget(&id.to_string());
            }
        }
        Ok(true)
    }

    /// Vertical right-sizing pass (Phase 3).
    ///
    /// For each running replica, observes usage into the [`VpaEngine`] and, in
    /// `Auto` mode, applies the recommendation via
    /// [`Runtime::update_container_resources`] (subject to the deadband). In
    /// `Recommend` mode it only logs. An `Unsupported` runtime triggers a real
    /// rolling restart; a `NotFound` container is skipped.
    async fn evaluate_vertical(&self, service_name: &str) -> Result<()> {
        let spec = {
            let specs = self.vertical_specs.read().await;
            match specs.get(service_name) {
                Some(s) => s.clone(),
                None => return Ok(()), // Vertical not enabled for this service.
            }
        };

        let containers = self
            .service_manager
            .read()
            .await
            .get_service_containers(service_name)
            .await;
        if containers.is_empty() {
            return Ok(());
        }

        // Observe every replica and pick the busiest recommendation: sizing to
        // the hottest replica avoids starving any single one.
        let mut chosen: Option<VpaRecommendation> = None;
        {
            let mut vpa = self.vpa.write().await;
            for id in &containers {
                let metrics_id = MetricsContainerId {
                    service: id.service.clone(),
                    replica: id.replica,
                };
                match self.stats_provider.get_stats(&metrics_id).await {
                    Ok(stats) => {
                        vpa.engine.observe(&id.to_string(), &stats);
                        if let Some(rec) = vpa.engine.recommend(&id.to_string(), &spec) {
                            chosen = Some(match chosen {
                                Some(c) => VpaRecommendation {
                                    cpu_millis: c.cpu_millis.max(rec.cpu_millis),
                                    memory_mib: c.memory_mib.max(rec.memory_mib),
                                },
                                None => rec,
                            });
                        }
                    }
                    Err(e) => debug!(
                        service = service_name,
                        container = %id,
                        error = %e,
                        "vertical: no stats for replica; skipping"
                    ),
                }
            }
        }

        let Some(rec) = chosen else {
            // Not enough samples yet to recommend anything.
            return Ok(());
        };

        match spec.mode {
            VerticalMode::Off => Ok(()),
            VerticalMode::Recommend => {
                info!(
                    service = service_name,
                    cpu_millis = rec.cpu_millis,
                    memory_mib = rec.memory_mib,
                    "vertical recommendation (recommend mode; not applied)"
                );
                Ok(())
            }
            VerticalMode::Auto => {
                Box::pin(self.apply_vertical(service_name, rec, &containers)).await
            }
        }
    }

    /// Apply a vertical recommendation in `Auto` mode, honoring the deadband
    /// and falling back to a rolling restart when the runtime cannot live-update.
    async fn apply_vertical(
        &self,
        service_name: &str,
        rec: VpaRecommendation,
        containers: &[ContainerId],
    ) -> Result<()> {
        // Deadband: skip if the recommendation barely moved.
        let prev = self
            .vpa
            .read()
            .await
            .last_applied
            .get(service_name)
            .copied();
        if !outside_deadband(prev, rec) {
            debug!(
                service = service_name,
                cpu_millis = rec.cpu_millis,
                memory_mib = rec.memory_mib,
                "vertical recommendation within deadband; skipping"
            );
            return Ok(());
        }

        let update = resource_update_for(rec);
        let mut needs_rolling_restart = false;

        for id in containers {
            match self.runtime.update_container_resources(id, &update).await {
                Ok(outcome) => {
                    if !outcome.warnings.is_empty() {
                        debug!(
                            service = service_name,
                            container = %id,
                            warnings = ?outcome.warnings,
                            "vertical apply produced warnings"
                        );
                    }
                }
                Err(AgentError::Unsupported(reason)) => {
                    debug!(
                        service = service_name,
                        container = %id,
                        reason = %reason,
                        "runtime cannot live-update resources; will roll the service"
                    );
                    needs_rolling_restart = true;
                    break;
                }
                Err(AgentError::NotFound { .. }) => {
                    debug!(
                        service = service_name,
                        container = %id,
                        "vertical apply: container vanished; skipping"
                    );
                }
                Err(e) => return Err(e),
            }
        }

        if needs_rolling_restart {
            Box::pin(self.rolling_restart_with_resources(service_name, rec)).await?;
        }

        info!(
            service = service_name,
            cpu_millis = rec.cpu_millis,
            memory_mib = rec.memory_mib,
            rolled = needs_rolling_restart,
            "applied vertical recommendation"
        );
        self.vpa
            .write()
            .await
            .last_applied
            .insert(service_name.to_string(), rec);
        Ok(())
    }

    /// Recreate a service's replicas one at a time, picking up `rec`'s
    /// resources, when the runtime cannot live-update a running container.
    ///
    /// REAL rolling restart: with a registered base [`ServiceSpec`] template
    /// (see [`AutoscaleController::set_service_template`]) we mutate its
    /// `resources` to the new CPU/memory and recreate each replica
    /// `stop → remove → create → start` individually, so the service never
    /// drops to zero capacity. Without a template we fall back to a
    /// `scale 0 → scale n` bounce (the recreated replicas then inherit the
    /// service's stored spec; resources are applied on the next live-update
    /// attempt where supported).
    async fn rolling_restart_with_resources(
        &self,
        service_name: &str,
        rec: VpaRecommendation,
    ) -> Result<()> {
        let template = self
            .service_templates
            .read()
            .await
            .get(service_name)
            .cloned();
        let containers = self
            .service_manager
            .read()
            .await
            .get_service_containers(service_name)
            .await;

        let Some(mut spec) = template else {
            // No template: bounce the whole service. Recreating replicas is a
            // real restart even if we can't re-create them one-by-one with the
            // exact runtime resources here.
            warn!(
                service = service_name,
                "rolling restart without a base ServiceSpec template; bouncing the service \
                 (call set_service_template to enable one-at-a-time recreation)"
            );
            let count = u32::try_from(containers.len()).unwrap_or(u32::MAX);
            {
                let sm = self.service_manager.read().await;
                sm.scale_service(service_name, 0).await?;
                sm.scale_service(service_name, count).await?;
            }
            self.record_scale_action(service_name).await;
            return Ok(());
        };

        // Mutate the template's resources to the recommendation. CPU is in
        // cores (millis/1000); memory is a Docker-style size string.
        let mut resources = spec.resources.clone();
        resources.cpu = Some(f64::from(rec.cpu_millis) / 1000.0);
        resources.memory = Some(format!("{}Mi", rec.memory_mib));
        spec.resources = resources;

        // Recreate replicas one at a time so the service keeps serving.
        for id in &containers {
            info!(
                service = service_name,
                container = %id,
                cpu_millis = rec.cpu_millis,
                memory_mib = rec.memory_mib,
                "rolling restart: recreating replica with new resources"
            );
            // Best-effort teardown of the old container, then recreate with the
            // resized spec. Errors on an individual replica are logged and the
            // roll continues so one bad replica doesn't strand the rest.
            if let Err(e) = self
                .runtime
                .stop_container(id, Duration::from_secs(10))
                .await
            {
                debug!(service = service_name, container = %id, error = %e, "rolling restart: stop failed (continuing)");
            }
            if let Err(e) = self.runtime.remove_container(id).await {
                debug!(service = service_name, container = %id, error = %e, "rolling restart: remove failed (continuing)");
            }
            self.vpa.write().await.engine.forget(&id.to_string());

            if let Err(e) = self.runtime.create_container(id, &spec).await {
                error!(service = service_name, container = %id, error = %e, "rolling restart: recreate failed");
                return Err(e);
            }
            if let Err(e) = self.runtime.start_container(id).await {
                error!(service = service_name, container = %id, error = %e, "rolling restart: start failed");
                return Err(e);
            }
        }
        self.record_scale_action(service_name).await;
        Ok(())
    }

    /// Evaluate a single service and execute scaling if needed
    async fn evaluate_and_scale(&self, service_name: &str) -> Result<()> {
        // Check cooldown first
        if !self.should_scale(service_name).await {
            return Ok(());
        }

        // Collect metrics
        let aggregated = match self.metrics.collect(service_name).await {
            Ok(m) => m,
            Err(e) => {
                // Missing metrics is not necessarily an error - the service might
                // not have any running containers yet
                debug!(
                    service = service_name,
                    error = %e,
                    "No metrics available for service"
                );
                return Ok(());
            }
        };

        // Make scaling decision
        let decision = {
            let mut autoscaler = self.autoscaler.write().await;
            match autoscaler.evaluate(service_name, &aggregated) {
                Ok(d) => d,
                Err(e) => {
                    debug!(
                        service = service_name,
                        error = %e,
                        "Failed to evaluate scaling"
                    );
                    return Ok(());
                }
            }
        };

        debug!(
            service = service_name,
            ?decision,
            cpu = aggregated.avg_cpu_percent,
            memory = aggregated.avg_memory_percent,
            instances = aggregated.instance_count,
            "Autoscale evaluation"
        );

        // Execute scaling if needed
        if let Some(target) = decision.target_replicas() {
            info!(
                service = service_name,
                target_replicas = target,
                decision = ?decision,
                "Executing autoscale"
            );

            // Execute the scaling
            if let Err(e) = self
                .service_manager
                .read()
                .await
                .scale_service(service_name, target)
                .await
            {
                error!(
                    service = service_name,
                    target = target,
                    error = %e,
                    "Failed to scale service"
                );
                return Err(e);
            }

            // A non-zero scale event counts as activity so the idle reaper
            // doesn't immediately tear back down what the autoscaler grew.
            if target > 0 {
                self.last_active
                    .write()
                    .await
                    .insert(service_name.to_string(), Instant::now());
            }

            // Record the scale action
            self.record_scale_action(service_name).await;

            // Update the autoscaler's internal state
            {
                let mut autoscaler = self.autoscaler.write().await;
                if let Err(e) = autoscaler.record_scale_action(service_name, target) {
                    warn!(
                        service = service_name,
                        error = %e,
                        "Failed to record scale action in autoscaler"
                    );
                }
            }
        }

        Ok(())
    }

    /// Signal shutdown of the autoscale loop
    pub fn shutdown(&self) {
        self.shutdown.notify_one();
    }

    /// Get the current evaluation interval
    #[must_use]
    pub fn interval(&self) -> Duration {
        self.interval
    }

    /// Get registered service count
    pub async fn registered_service_count(&self) -> usize {
        let specs = self.service_specs.read().await;
        specs.len()
    }
}

/// Check if any service in a deployment has adaptive scaling
///
/// This is a helper function to determine if the autoscale controller should
/// be started for a deployment.
#[must_use]
#[allow(clippy::implicit_hasher)]
pub fn has_adaptive_scaling(services: &HashMap<String, zlayer_spec::ServiceSpec>) -> bool {
    services
        .values()
        .any(|s| matches!(s.scale, ScaleSpec::Adaptive { .. }))
}

#[cfg(test)]
#[allow(deprecated)]
mod tests {
    use super::*;
    use crate::runtime::MockRuntime;
    use zlayer_scheduler::metrics::{MockMetricsSource, ServiceMetrics};
    use zlayer_spec::ScaleTargets;

    fn mock_spec() -> zlayer_spec::ServiceSpec {
        serde_yaml::from_str::<zlayer_spec::DeploymentSpec>(
            r"
version: v1
deployment: test
services:
  test:
    rtype: service
    image:
      name: test:latest
    endpoints:
      - name: http
        protocol: http
        port: 8080
    scale:
      mode: fixed
      replicas: 1
",
        )
        .unwrap()
        .services
        .remove("test")
        .unwrap()
    }

    fn adaptive_spec(
        min: u32,
        max: u32,
        cpu_target: Option<u8>,
        memory_target: Option<u8>,
    ) -> ScaleSpec {
        ScaleSpec::Adaptive {
            min,
            max,
            cooldown: Some(Duration::from_secs(0)), // No cooldown for tests
            targets: ScaleTargets {
                cpu: cpu_target,
                memory: memory_target,
                rps: None,
                custom: Vec::new(),
                external: Vec::new(),
            },
            behavior: None,
            triggers: Vec::new(),
            idle_window: None,
            vertical: None,
            predictive: None,
        }
    }

    fn raw_stats(cpu_usec: u64, mem_bytes: u64) -> RawContainerStats {
        RawContainerStats {
            cpu_usage_usec: cpu_usec,
            memory_bytes: mem_bytes,
            memory_limit: 512 * 1024 * 1024,
            timestamp: Instant::now(),
        }
    }

    #[test]
    fn test_vpa_percentile_nearest_rank() {
        let mut samples = std::collections::VecDeque::new();
        for v in [10.0, 20.0, 30.0, 40.0, 50.0] {
            samples.push_back(v);
        }
        assert_eq!(ContainerUsageHistory::percentile(&samples, 100), Some(50.0));
        assert_eq!(ContainerUsageHistory::percentile(&samples, 0), Some(10.0));
        // p90 nearest-rank of 5 samples => index ceil(0.9*5)-1 = 4 => 50.0
        assert_eq!(ContainerUsageHistory::percentile(&samples, 90), Some(50.0));
    }

    #[test]
    fn test_vpa_recommend_clamps_to_bounds() {
        let mut engine = VpaEngine::new();
        let id = "svc-rep-1";
        // Two samples with ZERO CPU delta: the container is idle, so the derived
        // rate is 0 millicores regardless of the wall-clock gap between samples
        // (deterministic — no flaky timing dependence). Memory holds steady at
        // 300 MiB. The recommendation should therefore clamp CPU *up* to the
        // 500 floor and memory *down* to the 256 ceiling.
        engine.observe(id, &raw_stats(1_000_000, 300 * 1024 * 1024));
        std::thread::sleep(Duration::from_millis(5));
        engine.observe(id, &raw_stats(1_000_000, 300 * 1024 * 1024));

        let spec = VerticalScaleSpec {
            mode: VerticalMode::Auto,
            min_cpu_millis: Some(500),
            max_cpu_millis: Some(2000),
            min_memory_mib: Some(128),
            max_memory_mib: Some(256),
            percentile: 90,
        };
        let rec = engine.recommend(id, &spec).expect("recommendation");
        // CPU clamped up to the 500 floor (idle => 0 raw); memory clamped down to
        // the 256 ceiling (300 MiB raw).
        assert_eq!(rec.cpu_millis, 500);
        assert_eq!(rec.memory_mib, 256);
    }

    #[test]
    fn test_deadband() {
        let base = VpaRecommendation {
            cpu_millis: 1000,
            memory_mib: 512,
        };
        // No previous => always apply.
        assert!(outside_deadband(None, base));
        // 5% change => within deadband.
        assert!(!outside_deadband(
            Some(base),
            VpaRecommendation {
                cpu_millis: 1050,
                memory_mib: 512
            }
        ));
        // 20% change => outside.
        assert!(outside_deadband(
            Some(base),
            VpaRecommendation {
                cpu_millis: 1200,
                memory_mib: 512
            }
        ));
    }

    #[test]
    fn test_resource_update_for() {
        let rec = VpaRecommendation {
            cpu_millis: 1500,
            memory_mib: 256,
        };
        let update = resource_update_for(rec);
        assert_eq!(update.cpu_period, Some(100_000));
        // 1500 millicores over a 100ms period => 150ms quota.
        assert_eq!(update.cpu_quota, Some(150_000));
        assert_eq!(update.memory, Some(256 * 1024 * 1024));
    }

    /// Wrap a freshly-built `ServiceManager` in the `Arc<RwLock<_>>` shape the
    /// controller now consumes (mirrors the daemon's post-`try_unwrap` handle).
    fn locked(runtime: &Arc<dyn Runtime + Send + Sync>) -> Arc<RwLock<ServiceManager>> {
        Arc::new(RwLock::new(ServiceManager::new(runtime.clone())))
    }

    #[tokio::test]
    async fn test_autoscale_controller_creation() {
        let runtime: Arc<dyn Runtime + Send + Sync> = Arc::new(MockRuntime::new());
        let manager = locked(&runtime);

        let controller = AutoscaleController::new(manager, runtime, Duration::from_secs(10));

        assert_eq!(controller.interval(), Duration::from_secs(10));
        assert_eq!(controller.registered_service_count().await, 0);
    }

    #[tokio::test]
    async fn test_register_service() {
        let runtime: Arc<dyn Runtime + Send + Sync> = Arc::new(MockRuntime::new());
        let manager = locked(&runtime);

        let controller = AutoscaleController::new(manager, runtime, Duration::from_secs(10));

        // Register an adaptive service
        let spec = adaptive_spec(1, 10, Some(70), None);
        controller.register_service("api", &spec, 2).await;

        assert!(controller.is_registered("api").await);
        assert_eq!(controller.registered_service_count().await, 1);
    }

    #[tokio::test]
    async fn test_register_fixed_service_ignored() {
        let runtime: Arc<dyn Runtime + Send + Sync> = Arc::new(MockRuntime::new());
        let manager = locked(&runtime);

        let controller = AutoscaleController::new(manager, runtime, Duration::from_secs(10));

        // Try to register a fixed service - should be ignored
        let spec = ScaleSpec::Fixed { replicas: 3 };
        controller.register_service("api", &spec, 3).await;

        assert!(!controller.is_registered("api").await);
        assert_eq!(controller.registered_service_count().await, 0);
    }

    #[tokio::test]
    async fn test_unregister_service() {
        let runtime: Arc<dyn Runtime + Send + Sync> = Arc::new(MockRuntime::new());
        let manager = locked(&runtime);

        let controller = AutoscaleController::new(manager, runtime, Duration::from_secs(10));

        let spec = adaptive_spec(1, 10, Some(70), None);
        controller.register_service("api", &spec, 2).await;

        assert!(controller.is_registered("api").await);

        controller.unregister_service("api").await;

        assert!(!controller.is_registered("api").await);
        assert_eq!(controller.registered_service_count().await, 0);
    }

    #[tokio::test]
    async fn test_has_adaptive_scaling() {
        let mut services = HashMap::new();

        // Add a fixed service
        let mut fixed_spec = mock_spec();
        fixed_spec.scale = ScaleSpec::Fixed { replicas: 3 };
        services.insert("web".to_string(), fixed_spec);

        // No adaptive services yet
        assert!(!has_adaptive_scaling(&services));

        // Add an adaptive service
        let mut adaptive = mock_spec();
        adaptive.scale = adaptive_spec(1, 10, Some(70), None);
        services.insert("api".to_string(), adaptive);

        // Now has adaptive
        assert!(has_adaptive_scaling(&services));
    }

    #[tokio::test]
    async fn test_autoscale_controller_with_mock_metrics() {
        let runtime: Arc<dyn Runtime + Send + Sync> = Arc::new(MockRuntime::new());
        let manager = locked(&runtime);

        // Create mock metrics source
        let mock = Arc::new(MockMetricsSource::new());

        // Set high CPU metrics
        mock.set_metrics(
            "api",
            vec![
                ServiceMetrics {
                    cpu_percent: 85.0,
                    memory_bytes: 100 * 1024 * 1024,
                    memory_limit: 512 * 1024 * 1024,
                    rps: None,
                    timestamp: Some(Instant::now()),
                    ..Default::default()
                },
                ServiceMetrics {
                    cpu_percent: 90.0,
                    memory_bytes: 150 * 1024 * 1024,
                    memory_limit: 512 * 1024 * 1024,
                    rps: None,
                    timestamp: Some(Instant::now()),
                    ..Default::default()
                },
            ],
        )
        .await;

        // Create controller with custom metrics
        let mut metrics = MetricsCollector::new();
        metrics.add_source(mock);

        let controller = AutoscaleController::with_custom_metrics(
            manager.clone(),
            runtime,
            metrics,
            Duration::from_secs(10),
        );

        // Register service
        Box::pin(
            manager
                .read()
                .await
                .upsert_service("api".to_string(), mock_spec()),
        )
        .await
        .unwrap();
        manager.read().await.scale_service("api", 2).await.unwrap();

        let spec = adaptive_spec(1, 10, Some(70), None);
        controller.register_service("api", &spec, 2).await;

        // Evaluate - should want to scale up due to high CPU
        controller.evaluate_and_scale("api").await.unwrap();

        // Check that scale happened (from 2 to 3)
        let count = manager
            .read()
            .await
            .service_replica_count("api")
            .await
            .unwrap();
        assert_eq!(count, 3);
    }

    #[tokio::test]
    async fn test_autoscale_controller_cooldown() {
        let runtime: Arc<dyn Runtime + Send + Sync> = Arc::new(MockRuntime::new());
        let manager = locked(&runtime);

        let controller = AutoscaleController::new(manager, runtime, Duration::from_secs(10));

        // Use a spec with 1 second cooldown
        let spec = ScaleSpec::Adaptive {
            min: 1,
            max: 10,
            cooldown: Some(Duration::from_secs(60)), // Long cooldown
            targets: ScaleTargets {
                cpu: Some(70),
                memory: None,
                rps: None,
                custom: Vec::new(),
                external: Vec::new(),
            },
            behavior: None,
            triggers: Vec::new(),
            idle_window: None,
            vertical: None,
            predictive: None,
        };

        controller.register_service("api", &spec, 2).await;

        // Initially should be able to scale
        assert!(controller.should_scale("api").await);

        // Record a scale action
        controller.record_scale_action("api").await;

        // Now should be in cooldown
        assert!(!controller.should_scale("api").await);
    }

    #[tokio::test]
    async fn test_scale_to_zero_after_idle_window() {
        let runtime: Arc<dyn Runtime + Send + Sync> = Arc::new(MockRuntime::new());
        let manager = locked(&runtime);

        let controller =
            AutoscaleController::new(manager.clone(), runtime, Duration::from_secs(10));

        // Stand up a service with two replicas.
        Box::pin(
            manager
                .read()
                .await
                .upsert_service("api".to_string(), mock_spec()),
        )
        .await
        .unwrap();
        manager.read().await.scale_service("api", 2).await.unwrap();

        // min == 0 + a tiny idle window so scale-to-zero is eligible.
        let spec = ScaleSpec::Adaptive {
            min: 0,
            max: 10,
            cooldown: Some(Duration::from_secs(0)),
            targets: ScaleTargets::default(),
            behavior: None,
            triggers: Vec::new(),
            idle_window: Some(Duration::from_millis(10)),
            vertical: None,
            predictive: None,
        };
        controller.register_service("api", &spec, 2).await;

        // Force the idle clock well into the past so the window has elapsed.
        controller.last_active.write().await.insert(
            "api".to_string(),
            Instant::now().checked_sub(Duration::from_secs(60)).unwrap(),
        );

        // First eval reaps to zero (MockRuntime stats are low CPU, so no
        // activity bump fights the reaper).
        let reaped = controller.evaluate_idle("api").await.unwrap();
        assert!(reaped, "service should have been reaped to zero");
        assert_eq!(
            manager
                .read()
                .await
                .service_replica_count("api")
                .await
                .unwrap(),
            0
        );
    }

    #[tokio::test]
    async fn test_mark_active_resets_idle_clock() {
        let runtime: Arc<dyn Runtime + Send + Sync> = Arc::new(MockRuntime::new());
        let manager = locked(&runtime);

        let controller =
            AutoscaleController::new(manager.clone(), runtime, Duration::from_secs(10));

        Box::pin(
            manager
                .read()
                .await
                .upsert_service("api".to_string(), mock_spec()),
        )
        .await
        .unwrap();
        manager.read().await.scale_service("api", 1).await.unwrap();

        let spec = ScaleSpec::Adaptive {
            min: 0,
            max: 10,
            cooldown: Some(Duration::from_secs(0)),
            targets: ScaleTargets::default(),
            behavior: None,
            triggers: Vec::new(),
            idle_window: Some(Duration::from_secs(300)),
            vertical: None,
            predictive: None,
        };
        controller.register_service("api", &spec, 1).await;

        // Push the clock back, then mark active to reset it.
        controller.last_active.write().await.insert(
            "api".to_string(),
            Instant::now()
                .checked_sub(Duration::from_secs(600))
                .unwrap(),
        );
        controller.mark_active_async("api").await;

        // Not idle anymore => not reaped.
        let reaped = controller.evaluate_idle("api").await.unwrap();
        assert!(!reaped, "marked-active service must not be reaped");
        assert_eq!(
            manager
                .read()
                .await
                .service_replica_count("api")
                .await
                .unwrap(),
            1
        );
    }

    #[tokio::test]
    async fn test_no_scale_to_zero_when_min_nonzero() {
        let runtime: Arc<dyn Runtime + Send + Sync> = Arc::new(MockRuntime::new());
        let manager = locked(&runtime);

        let controller =
            AutoscaleController::new(manager.clone(), runtime, Duration::from_secs(10));

        Box::pin(
            manager
                .read()
                .await
                .upsert_service("api".to_string(), mock_spec()),
        )
        .await
        .unwrap();
        manager.read().await.scale_service("api", 2).await.unwrap();

        // min == 1 disables scale-to-zero even with an elapsed idle window.
        let spec = ScaleSpec::Adaptive {
            min: 1,
            max: 10,
            cooldown: Some(Duration::from_secs(0)),
            targets: ScaleTargets::default(),
            behavior: None,
            triggers: Vec::new(),
            idle_window: Some(Duration::from_millis(1)),
            vertical: None,
            predictive: None,
        };
        controller.register_service("api", &spec, 2).await;
        controller.last_active.write().await.insert(
            "api".to_string(),
            Instant::now().checked_sub(Duration::from_secs(60)).unwrap(),
        );

        let reaped = controller.evaluate_idle("api").await.unwrap();
        assert!(!reaped, "min>0 must never scale to zero");
        assert_eq!(
            manager
                .read()
                .await
                .service_replica_count("api")
                .await
                .unwrap(),
            2
        );
    }

    #[tokio::test]
    async fn test_autoscale_controller_shutdown() {
        let runtime: Arc<dyn Runtime + Send + Sync> = Arc::new(MockRuntime::new());
        let manager = locked(&runtime);

        let controller = Arc::new(AutoscaleController::new(
            manager,
            runtime,
            Duration::from_millis(100), // Fast interval for test
        ));

        let controller_clone = controller.clone();

        // Spawn the loop (boxed: run_loop's future is large — clippy::large_future).
        let handle = tokio::spawn(async move { Box::pin(controller_clone.run_loop()).await });

        // Let it run briefly
        tokio::time::sleep(Duration::from_millis(50)).await;

        // Signal shutdown
        controller.shutdown();

        // Should complete without error
        let result = handle.await.unwrap();
        assert!(result.is_ok());
    }
}