//! Process hypervisor - spawns and supervises instances

use crate::cgroup::{CgroupManager, ResourceLimits};
use crate::config::Config;
use crate::instance::{HealthStatus, Instance, InstanceId, InstanceInfo};
use crate::logs::LogBuffer;
use crate::metrics::Metrics;
use crate::port_allocator::PortAllocator;
use crate::runtime::{NamespaceRuntime, ProcessRuntime, Runtime, RuntimeHandle, RuntimeType, SpawnConfig};
use crate::storage::{calculate_dir_size, StorageInfo};
#[cfg(feature = "sandbox")]
use crate::runtime::SandboxRuntime;
use anyhow::{Context, Result};
use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::io::{AsyncBufReadExt, BufReader};
use tokio::sync::RwLock;
use tracing::{error, info, warn};

const HEALTH_CHECK_TIMEOUT: Duration = Duration::from_secs(5);

/// RAII guard that decrements the active connection count when dropped.
pub struct ConnectionGuard {
    counter: Arc<std::sync::atomic::AtomicU32>,
}

impl Drop for ConnectionGuard {
    fn drop(&mut self) {
        self.counter.fetch_sub(1, std::sync::atomic::Ordering::Relaxed);
    }
}

/// The hypervisor manages all running instances
pub struct Hypervisor {
    config: Config,
    instances: RwLock<HashMap<InstanceId, Instance>>,
    /// Guard against concurrent spawns of the same instance.
    /// An instance ID is added before spawn begins and removed after it completes.
    spawning: RwLock<std::collections::HashSet<InstanceId>>,
    /// Wake-once notifications: when an instance is being woken, other requests
    /// wait on the Notify instead of spawning duplicate processes.
    waking: RwLock<HashMap<InstanceId, Arc<tokio::sync::Notify>>>,
    /// Active connection count per instance (for connection-aware idle timeout and draining)
    active_connections: RwLock<HashMap<InstanceId, Arc<std::sync::atomic::AtomicU32>>>,
    /// Restart history that persists across stop/spawn cycles.
    /// Maps instance ID to (restart_count, restart_times).
    restart_history: RwLock<HashMap<InstanceId, (u32, Vec<Instant>)>>,
    log_buffer: Arc<LogBuffer>,
    metrics: Arc<Metrics>,
    /// Port allocator for TCP ports (30000-40000)
    port_allocator: Arc<PortAllocator>,
    /// Process runtime (always available, fallback)
    process_runtime: ProcessRuntime,
    /// Namespace runtime (default on Linux)
    namespace_runtime: NamespaceRuntime,
    /// Sandbox runtime (gVisor) - requires runsc
    #[cfg(feature = "sandbox")]
    sandbox_runtime: SandboxRuntime,
    /// Cgroup manager for resource limits (Linux cgroups v2)
    cgroup_manager: CgroupManager,
    /// Optional state store for crash recovery persistence
    state_store: Option<Arc<crate::store::StateStore>>,
}

impl Hypervisor {
    /// Create a new hypervisor with the given config
    pub fn new(config: Config) -> Arc<Self> {
        let namespace_runtime = NamespaceRuntime::new();
        let cgroup_manager = CgroupManager::new();
        let port_allocator = Arc::new(PortAllocator::new());

        Arc::new(Self {
            config,
            instances: RwLock::new(HashMap::new()),
            spawning: RwLock::new(std::collections::HashSet::new()),
            waking: RwLock::new(HashMap::new()),
            active_connections: RwLock::new(HashMap::new()),
            restart_history: RwLock::new(HashMap::new()),
            log_buffer: LogBuffer::new(),
            metrics: Metrics::new(),
            port_allocator,
            process_runtime: ProcessRuntime::new(),
            namespace_runtime,
            #[cfg(feature = "sandbox")]
            sandbox_runtime: SandboxRuntime::new(),
            cgroup_manager,
            state_store: None,
        })
    }

    /// Create a new hypervisor with a custom log buffer
    pub fn with_log_buffer(config: Config, log_buffer: Arc<LogBuffer>) -> Arc<Self> {
        let namespace_runtime = NamespaceRuntime::new();
        let cgroup_manager = CgroupManager::new();
        let port_allocator = Arc::new(PortAllocator::new());

        Arc::new(Self {
            config,
            instances: RwLock::new(HashMap::new()),
            spawning: RwLock::new(std::collections::HashSet::new()),
            waking: RwLock::new(HashMap::new()),
            active_connections: RwLock::new(HashMap::new()),
            restart_history: RwLock::new(HashMap::new()),
            log_buffer,
            metrics: Metrics::new(),
            port_allocator,
            process_runtime: ProcessRuntime::new(),
            namespace_runtime,
            #[cfg(feature = "sandbox")]
            sandbox_runtime: SandboxRuntime::new(),
            cgroup_manager,
            state_store: None,
        })
    }

    /// Create a new hypervisor with a state store for crash recovery
    pub fn with_state_store(config: Config, state_store: Arc<crate::store::StateStore>) -> Arc<Self> {
        let mut hyp = Self::new(config);
        // SAFETY: Arc::get_mut works because we just created this Arc and hold the only reference
        Arc::get_mut(&mut hyp).expect("just created Arc").state_store = Some(state_store);
        hyp
    }

    /// Get the log buffer
    pub fn log_buffer(&self) -> Arc<LogBuffer> {
        self.log_buffer.clone()
    }

    /// Get the metrics
    pub fn metrics(&self) -> Arc<Metrics> {
        self.metrics.clone()
    }

    /// Load config from tenement.toml and create hypervisor
    pub fn from_config_file() -> Result<Arc<Self>> {
        let config = Config::load()?;
        Ok(Self::new(config))
    }

    /// Spawn a new instance of a process
    pub async fn spawn(&self, process_name: &str, id: &str) -> Result<PathBuf> {
        self.spawn_with_env(process_name, id, HashMap::new()).await
    }

    /// Spawn a new instance with additional environment variables
    pub async fn spawn_with_env(
        &self,
        process_name: &str,
        id: &str,
        extra_env: HashMap<String, String>,
    ) -> Result<PathBuf> {
        let process_config = self
            .config
            .get_service(process_name)
            .with_context(|| format!("Unknown process: {}", process_name))?
            .clone();

        let instance_id = InstanceId::new(process_name, id);
        let data_dir = &self.config.settings.data_dir;
        let socket = process_config.socket_path(process_name, id);

        // Create instance data directory
        let instance_data_dir = data_dir.join(process_name).join(id);
        std::fs::create_dir_all(&instance_data_dir)
            .with_context(|| format!("Failed to create data dir: {:?}", instance_data_dir))?;

        // Create socket parent directory if needed
        if let Some(socket_parent) = socket.parent() {
            std::fs::create_dir_all(socket_parent)
                .with_context(|| format!("Failed to create socket dir: {:?}", socket_parent))?;
        }

        // Atomically check if running/spawning and mark as spawning (prevents race condition).
        // Both checks and the insert happen under one write lock.
        {
            let instances = self.instances.read().await;
            if instances.contains_key(&instance_id) {
                info!("Instance {} already running", instance_id);
                return Ok(socket);
            }
            let mut spawning = self.spawning.write().await;
            if !spawning.insert(instance_id.clone()) {
                info!("Instance {} is already being spawned", instance_id);
                return Ok(socket);
            }
        }

        let data_dir = &self.config.settings.data_dir;

        // Validate isolation level is available - fail loudly if not
        let isolation = process_config.isolation;
        match isolation {
            RuntimeType::Namespace => {
                if !self.namespace_runtime.is_available() {
                    anyhow::bail!(
                        "Instance {}: namespace isolation requires Linux. \
                         Set isolation = \"process\" in your config for local development.",
                        instance_id
                    );
                }
            }
            RuntimeType::Process => {}
            RuntimeType::Sandbox => {
                #[cfg(feature = "sandbox")]
                {
                    if !self.sandbox_runtime.is_available() {
                        anyhow::bail!(
                            "Instance {}: sandbox isolation requires gVisor (runsc).\n\
                            Install: https://gvisor.dev/docs/user_guide/install/\n\
                            Or use isolation = \"namespace\" for trusted code.",
                            instance_id
                        );
                    }
                }
                #[cfg(not(feature = "sandbox"))]
                {
                    anyhow::bail!(
                        "Instance {}: sandbox isolation requires the 'sandbox' feature.\n\
                        Compile with: cargo build --features sandbox",
                        instance_id
                    );
                }
            }
            RuntimeType::Firecracker | RuntimeType::Qemu => {
                anyhow::bail!(
                    "Instance {}: {} isolation not yet supported in hypervisor",
                    instance_id, isolation
                );
            }
        }

        info!("Spawning instance {} (isolation: {})", instance_id, isolation);

        // Allocate a TCP port for process/namespace/sandbox runtimes
        // VMs (Firecracker/QEMU) use vsock, so they don't need TCP ports
        let port = match isolation {
            RuntimeType::Process | RuntimeType::Namespace | RuntimeType::Sandbox => {
                Some(self.port_allocator.allocate().await
                    .with_context(|| format!("Failed to allocate port for {}", instance_id))?)
            }
            RuntimeType::Firecracker | RuntimeType::Qemu => None,
        };

        // Build environment
        // If the user wrote `command = "uv run python app.py"` with no args,
        // shell-split the command string into executable + arguments.
        let raw_command = process_config.command_interpolated(process_name, id, data_dir, port);
        let explicit_args = process_config.args_interpolated(process_name, id, data_dir, port);
        let (command, args) = if explicit_args.is_empty() {
            let parts = shell_words::split(&raw_command)
                .with_context(|| format!("Failed to parse command: {}", raw_command))?;
            let (cmd, rest) = parts.split_first()
                .map(|t| (t.0.clone(), t.1.to_vec()))
                .unwrap_or((raw_command, vec![]));
            (cmd, rest)
        } else {
            (raw_command, explicit_args)
        };
        let mut env = process_config.env_interpolated(process_name, id, data_dir, port);

        // Merge extra env vars
        env.extend(extra_env);

        // Always set SOCKET_PATH for backwards compatibility and test scripts
        env.insert("SOCKET_PATH".to_string(), socket.to_string_lossy().to_string());

        // Also set PORT for TCP-based runtimes (Process/Namespace/Sandbox)
        if let Some(port) = port {
            env.insert("PORT".to_string(), port.to_string());
        }

        // Build spawn config
        let spawn_config = SpawnConfig {
            command,
            args,
            env,
            socket: socket.clone(),
            workdir: process_config.workdir.clone(),
            vm_config: None,
        };

        // Spawn using the selected isolation level (we already validated it's available above)
        let mut handle = match isolation {
            RuntimeType::Namespace => self.namespace_runtime.spawn(&spawn_config).await?,
            RuntimeType::Process => self.process_runtime.spawn(&spawn_config).await?,
            #[cfg(feature = "sandbox")]
            RuntimeType::Sandbox => self.sandbox_runtime.spawn(&spawn_config).await?,
            #[cfg(not(feature = "sandbox"))]
            RuntimeType::Sandbox => unreachable!("sandbox feature not enabled"),
            // Firecracker/Qemu already rejected above
            _ => unreachable!(),
        };

        // Apply resource limits via cgroups v2 (Linux only)
        let resource_limits = ResourceLimits {
            memory_limit_mb: process_config.memory_limit_mb,
            cpu_shares: process_config.cpu_shares,
        };
        if resource_limits.has_limits() {
            // Create cgroup and add process. Fail loudly if resource limits are
            // configured but can't be applied (process would run unrestricted).
            if let Err(e) = self.cgroup_manager.create_cgroup(&instance_id.to_string(), &resource_limits) {
                // Kill the already-spawned child and clean up spawning guard
                let _ = handle.kill().await;
                self.spawning.write().await.remove(&instance_id);
                return Err(e).with_context(|| format!(
                    "Failed to create cgroup for {}. Resource limits will not be enforced.", instance_id
                ));
            }

            if let Some(pid) = handle.pid() {
                if let Err(e) = self.cgroup_manager.add_process(&instance_id.to_string(), pid, &resource_limits) {
                    let _ = handle.kill().await;
                    self.spawning.write().await.remove(&instance_id);
                    return Err(e).with_context(|| format!(
                        "Failed to add process to cgroup for {}. Resource limits will not be enforced.", instance_id
                    ));
                }
            }
        }

        // Set up log capture for process-based runtimes (Process and Namespace both have child processes)
        match &mut handle {
            RuntimeHandle::Process { ref mut child, .. }
            | RuntimeHandle::Namespace { ref mut child, .. } => {
                // Take stdout/stderr handles and spawn capture tasks
                let stdout = child.stdout.take();
                let stderr = child.stderr.take();

                // Spawn stdout capture task
                if let Some(stdout) = stdout {
                    let log_buffer = self.log_buffer.clone();
                    let process = process_name.to_string();
                    let inst_id = id.to_string();
                    tokio::spawn(async move {
                        let reader = BufReader::new(stdout);
                        let mut lines = reader.lines();
                        while let Ok(Some(line)) = lines.next_line().await {
                            log_buffer.push_stdout(&process, &inst_id, line).await;
                        }
                    });
                }

                // Spawn stderr capture task
                if let Some(stderr) = stderr {
                    let log_buffer = self.log_buffer.clone();
                    let process = process_name.to_string();
                    let inst_id = id.to_string();
                    tokio::spawn(async move {
                        let reader = BufReader::new(stderr);
                        let mut lines = reader.lines();
                        while let Ok(Some(line)) = lines.next_line().await {
                            log_buffer.push_stderr(&process, &inst_id, line).await;
                        }
                    });
                }
            }
            _ => {
                // VM runtimes handle logging differently
            }
        }

        let runtime_type = handle.runtime_type();
        let now = Instant::now();

        // Restore restart history from persistent storage (survives stop/spawn cycles)
        let (restarts, restart_times) = {
            let history = self.restart_history.read().await;
            history.get(&instance_id).cloned().unwrap_or((0, Vec::new()))
        };

        let instance = Instance {
            id: instance_id.clone(),
            handle,
            runtime_type,
            socket: socket.clone(),
            port,
            started_at: now,
            restarts,
            consecutive_failures: 0,
            last_health_check: None,
            health_status: HealthStatus::Unknown,
            restart_times,
            last_activity: now,
            idle_timeout: process_config.idle_timeout,
            storage_quota_mb: process_config.storage_quota_mb,
            storage_persist: process_config.storage_persist,
            storage_used_bytes: 0,
            data_dir: instance_data_dir.clone(),
            weight: 100, // Default weight - receives full traffic
        };

        {
            let mut instances = self.instances.write().await;
            instances.insert(instance_id.clone(), instance);
        }

        // Remove from spawning set now that instance is registered
        {
            let mut spawning = self.spawning.write().await;
            spawning.remove(&instance_id);
        }

        // Update metrics
        self.metrics.instances_up.inc();

        // Persist instance state for crash recovery (only if we have a PID to track)
        if let Some(ref store) = self.state_store {
            let pid = {
                let instances = self.instances.read().await;
                instances.get(&instance_id).and_then(|i| i.handle.pid())
            };
            if let Some(pid) = pid {
                let state = crate::store::InstanceState {
                    instance_id: instance_id.to_string(),
                    process_name: process_name.to_string(),
                    id: id.to_string(),
                    pid,
                    port,
                    started_at: chrono::Utc::now().to_rfc3339(),
                };
                if let Err(e) = store.save(&state).await {
                    error!("Failed to persist instance state for {}: {}", instance_id, e);
                }
            }
        }

        // Spawn exit monitor: detects process exit within 1s instead of
        // waiting for the next health check cycle (up to 10s).
        if let Some(pid) = {
            let instances = self.instances.read().await;
            instances.get(&instance_id).and_then(|i| i.handle.pid())
        } {
            let exit_instance_id = instance_id.clone();
            let log_buffer = self.log_buffer.clone();
            // Reference to the instances map so the monitor can check
            // if the instance was intentionally stopped (removed from map).
            let instances_ref = unsafe {
                // SAFETY: The RwLock<HashMap> lives as long as the Arc<Hypervisor>,
                // which outlives all spawned instances.
                &*(&self.instances as *const RwLock<HashMap<InstanceId, Instance>>)
            };
            tokio::spawn(async move {
                loop {
                    tokio::time::sleep(Duration::from_secs(1)).await;

                    #[cfg(unix)]
                    let alive = unsafe { libc::kill(pid as i32, 0) } == 0;
                    #[cfg(not(unix))]
                    let alive = true;

                    if !alive {
                        // Check if instance was intentionally stopped (removed from map)
                        let still_tracked = {
                            let map = instances_ref.read().await;
                            map.contains_key(&exit_instance_id)
                        };
                        if still_tracked {
                            error!(
                                "Instance {} (pid {}) exited unexpectedly",
                                exit_instance_id, pid
                            );
                            log_buffer.push_stderr(
                                &exit_instance_id.process,
                                &exit_instance_id.id,
                                format!("Process exited unexpectedly (pid {})", pid),
                            ).await;
                        }
                        break;
                    }
                }
            });
        }

        // Wait for service to be ready
        if let Some(port) = port {
            // TCP mode: try to connect
            let addr = format!("127.0.0.1:{}", port);
            for _ in 0..50 {
                if tokio::net::TcpStream::connect(&addr).await.is_ok() {
                    info!("Instance {} ready at {}", instance_id, addr);
                    return Ok(socket);
                }
                tokio::time::sleep(Duration::from_millis(10)).await;
            }
            warn!("Instance {} TCP port {} not ready after 500ms", instance_id, port);
        } else {
            // Socket mode: check if file exists (VMs use vsock)
            for _ in 0..50 {
                if socket.exists() {
                    info!("Instance {} ready at {:?}", instance_id, socket);
                    return Ok(socket);
                }
                tokio::time::sleep(Duration::from_millis(10)).await;
            }
            warn!("Instance {} socket not ready after 500ms", instance_id);
        }

        Ok(socket)
    }

    /// Stop all running instances. Called on graceful shutdown.
    pub async fn stop_all(&self) {
        let instance_ids: Vec<InstanceId> = {
            let instances = self.instances.read().await;
            instances.keys().cloned().collect()
        };

        if instance_ids.is_empty() {
            return;
        }

        info!("Stopping {} instance(s) for shutdown", instance_ids.len());
        for instance_id in instance_ids {
            if let Err(e) = self.stop(&instance_id.process, &instance_id.id).await {
                error!("Failed to stop {} during shutdown: {}", instance_id, e);
            }
        }
        info!("All instances stopped");
    }

    /// Stop an instance. Waits up to 5 seconds for active connections to drain.
    pub async fn stop(&self, process_name: &str, id: &str) -> Result<()> {
        let instance_id = InstanceId::new(process_name, id);

        // Clear spawning guard if present (in case spawn failed and left it)
        {
            let mut spawning = self.spawning.write().await;
            spawning.remove(&instance_id);
        }

        // Wait for active connections to drain (up to 5 seconds)
        let active = self.active_connection_count(process_name, id).await;
        if active > 0 {
            info!(
                "Instance {} has {} active connection(s), draining...",
                instance_id, active
            );
            for _ in 0..50 {
                if self.active_connection_count(process_name, id).await == 0 {
                    break;
                }
                tokio::time::sleep(Duration::from_millis(100)).await;
            }
            let remaining = self.active_connection_count(process_name, id).await;
            if remaining > 0 {
                warn!(
                    "Instance {} still has {} connection(s) after drain timeout, force stopping",
                    instance_id, remaining
                );
            }
        }

        let mut instances = self.instances.write().await;

        if let Some(mut instance) = instances.remove(&instance_id) {
            info!("Stopping instance {}", instance_id);

            instance
                .handle
                .kill()
                .await
                .with_context(|| format!("Failed to kill process: {}", instance_id))?;

            // Release allocated port back to the pool
            if let Some(port) = instance.port {
                self.port_allocator.release(port).await;
            }

            // Clean up cgroup (if one was created)
            if let Err(e) = self.cgroup_manager.remove_cgroup(&instance_id.to_string()) {
                warn!("Failed to remove cgroup for {}: {}", instance_id, e);
            }

            // Clean up socket
            if instance.socket.exists() {
                std::fs::remove_file(&instance.socket).ok();
            }

            // Clean up data directory if storage_persist is false
            if !instance.storage_persist && instance.data_dir.exists() {
                if let Err(e) = std::fs::remove_dir_all(&instance.data_dir) {
                    warn!(
                        "Failed to remove data directory {:?} for {}: {}",
                        instance.data_dir, instance_id, e
                    );
                } else {
                    info!("Removed data directory {:?} for {}", instance.data_dir, instance_id);
                }
            }

            // Update metrics
            self.metrics.instances_up.dec();

            // Remove persisted state
            if let Some(ref store) = self.state_store {
                if let Err(e) = store.remove(&instance_id.to_string()).await {
                    error!("Failed to remove instance state for {}: {}", instance_id, e);
                }
            }

            Ok(())
        } else {
            anyhow::bail!("Instance not found: {}", instance_id)
        }
    }

    /// Restart an instance with exponential backoff
    pub async fn restart(&self, process_name: &str, id: &str) -> Result<PathBuf> {
        let instance_id = InstanceId::new(process_name, id);

        // Get restart count from persistent history (survives stop/spawn cycles)
        let restarts = {
            let history = self.restart_history.read().await;
            history.get(&instance_id).map(|(count, _)| *count).unwrap_or(0)
        };

        // Stop if running
        let _ = self.stop(process_name, id).await;

        // Calculate and apply exponential backoff delay
        let backoff_delay = self.calculate_backoff(restarts);
        if backoff_delay > Duration::ZERO {
            info!(
                "Applying backoff delay of {:?} before restarting {} (restart #{})",
                backoff_delay, instance_id, restarts + 1
            );
            tokio::time::sleep(backoff_delay).await;
        }

        // Spawn again
        let socket = self.spawn(process_name, id).await?;

        // Update persistent restart history
        let window = Duration::from_secs(self.config.settings.restart_window);
        {
            let mut history = self.restart_history.write().await;
            let entry = history.entry(instance_id.clone()).or_insert((0, Vec::new()));
            entry.0 = restarts + 1;
            entry.1.push(Instant::now());
            entry.1.retain(|t| t.elapsed() < window);
        }

        // Also update the instance's restart count for display
        {
            let mut instances = self.instances.write().await;
            if let Some(instance) = instances.get_mut(&instance_id) {
                instance.restarts = restarts + 1;
                // Copy restart_times from persistent history
                let history = self.restart_history.read().await;
                if let Some((_, times)) = history.get(&instance_id) {
                    instance.restart_times = times.clone();
                }
            }
        }

        // Update metrics
        let mut labels = HashMap::new();
        labels.insert("process".to_string(), process_name.to_string());
        labels.insert("id".to_string(), id.to_string());
        let counter = self.metrics.instance_restarts.with_labels(&labels).await;
        counter.inc();

        Ok(socket)
    }

    /// Calculate exponential backoff delay based on restart count
    /// Formula: base * 2^(restarts - 1), capped at max
    fn calculate_backoff(&self, restarts: u32) -> Duration {
        if restarts == 0 {
            return Duration::ZERO;
        }

        let base_ms = self.config.settings.backoff_base_ms;
        let max_ms = self.config.settings.backoff_max_ms;

        // Calculate delay: base * 2^(restarts - 1)
        // Cap the shift amount to prevent overflow (max 63 bits for u64)
        let shift = (restarts - 1).min(63);
        let multiplier = 1u64.checked_shl(shift).unwrap_or(u64::MAX);
        let delay_ms = base_ms.saturating_mul(multiplier).min(max_ms);

        Duration::from_millis(delay_ms)
    }

    /// Check if an instance is running
    pub async fn is_running(&self, process_name: &str, id: &str) -> bool {
        let instance_id = InstanceId::new(process_name, id);
        let instances = self.instances.read().await;
        instances.contains_key(&instance_id)
    }

    /// Spawn if not already running
    pub async fn spawn_if_not_running(&self, process_name: &str, id: &str) -> Result<PathBuf> {
        if self.is_running(process_name, id).await {
            let process_config = self
                .config
                .get_service(process_name)
                .context("Unknown process")?;
            Ok(process_config.socket_path(process_name, id))
        } else {
            self.spawn(process_name, id).await
        }
    }

    /// List all running instances
    pub async fn list(&self) -> Vec<InstanceInfo> {
        let instances = self.instances.read().await;
        instances.values().map(|i| i.info()).collect()
    }

    /// Get info for a specific instance
    pub async fn get(&self, process_name: &str, id: &str) -> Option<InstanceInfo> {
        let instance_id = InstanceId::new(process_name, id);
        let instances = self.instances.read().await;
        instances.get(&instance_id).map(|i| i.info())
    }

    /// Get storage information for a specific instance
    pub async fn get_storage_info(&self, process_name: &str, id: &str) -> Option<StorageInfo> {
        let instance_id = InstanceId::new(process_name, id);
        let (data_dir, quota_mb) = {
            let instances = self.instances.read().await;
            instances.get(&instance_id).map(|i| {
                (i.data_dir.clone(), i.storage_quota_mb)
            })?
        };

        // Calculate current directory size
        let used_bytes = calculate_dir_size(data_dir.clone()).await.unwrap_or(0);

        // Convert quota from MB to bytes
        let quota_bytes = quota_mb.map(|mb| (mb as u64) * 1024 * 1024);

        Some(StorageInfo::new(used_bytes, quota_bytes, data_dir))
    }

    /// Get instance info and touch activity atomically.
    /// This prevents race conditions where an instance could be reaped
    /// between checking if it's running and touching its activity.
    pub async fn get_and_touch(&self, process_name: &str, id: &str) -> Option<InstanceInfo> {
        let instance_id = InstanceId::new(process_name, id);
        let mut instances = self.instances.write().await;
        if let Some(instance) = instances.get_mut(&instance_id) {
            instance.touch();
            Some(instance.info())
        } else {
            None
        }
    }

    /// Check if a process is configured (can be spawned)
    pub fn has_process(&self, process_name: &str) -> bool {
        self.config.get_service(process_name).is_some()
    }

    /// Increment active connection count for an instance. Returns a guard
    /// that decrements the count when dropped.
    pub async fn connection_start(&self, process_name: &str, id: &str) -> ConnectionGuard {
        let instance_id = InstanceId::new(process_name, id);
        let counter = {
            let mut conns = self.active_connections.write().await;
            conns
                .entry(instance_id)
                .or_insert_with(|| Arc::new(std::sync::atomic::AtomicU32::new(0)))
                .clone()
        };
        counter.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        ConnectionGuard { counter }
    }

    /// Get active connection count for an instance
    pub async fn active_connection_count(&self, process_name: &str, id: &str) -> u32 {
        let instance_id = InstanceId::new(process_name, id);
        let conns = self.active_connections.read().await;
        conns
            .get(&instance_id)
            .map(|c| c.load(std::sync::atomic::Ordering::Relaxed))
            .unwrap_or(0)
    }

    /// Get the request timeout for a process (in seconds)
    pub fn request_timeout(&self, process_name: &str) -> Duration {
        let secs = self
            .config
            .get_service(process_name)
            .map(|p| p.request_timeout)
            .unwrap_or(30);
        Duration::from_secs(secs)
    }

    /// Check health of an instance
    pub async fn check_health(&self, process_name: &str, id: &str) -> HealthStatus {
        let instance_id = InstanceId::new(process_name, id);

        let process_config = match self.config.get_service(process_name) {
            Some(c) => c,
            None => return HealthStatus::Unknown,
        };

        // If no health endpoint configured, assume healthy if socket exists
        let health_endpoint = match &process_config.health {
            Some(h) => h,
            None => {
                let socket = process_config.socket_path(process_name, id);
                return if socket.exists() {
                    HealthStatus::Healthy
                } else {
                    HealthStatus::Unhealthy
                };
            }
        };

        // Get socket, vsock port, and TCP port from the running instance
        let (socket, vsock_port, tcp_port) = {
            let instances = self.instances.read().await;
            match instances.get(&instance_id) {
                Some(instance) => (
                    instance.handle.socket().clone(),
                    instance.handle.vsock_port(),
                    instance.port,
                ),
                None => return HealthStatus::Unknown,
            }
        };

        // Use TCP health check for process/namespace/sandbox runtimes,
        // fall back to Unix socket for VMs
        let result = if let Some(port) = tcp_port {
            self.ping_health_tcp(port, health_endpoint).await
        } else {
            self.ping_health_with_vsock(&socket, health_endpoint, vsock_port).await
        };

        let mut instances = self.instances.write().await;
        let instance = match instances.get_mut(&instance_id) {
            Some(i) => i,
            None => return HealthStatus::Unknown,
        };

        instance.last_health_check = Some(Instant::now());

        match result {
            Ok(()) => {
                instance.consecutive_failures = 0;
                instance.health_status = HealthStatus::Healthy;
                HealthStatus::Healthy
            }
            Err(e) => {
                instance.consecutive_failures += 1;
                warn!(
                    "Health check failed for {}: {} (failures: {})",
                    instance_id, e, instance.consecutive_failures
                );

                let status = match instance.consecutive_failures {
                    1..=2 => HealthStatus::Degraded,
                    _ => {
                        let window = Duration::from_secs(self.config.settings.restart_window);
                        let recent_restarts = instance
                            .restart_times
                            .iter()
                            .filter(|t| t.elapsed() < window)
                            .count() as u32;

                        if recent_restarts >= self.config.settings.max_restarts {
                            HealthStatus::Failed
                        } else {
                            HealthStatus::Unhealthy
                        }
                    }
                };
                instance.health_status = status;
                status
            }
        }
    }

    /// Ping a health endpoint via TCP (for process/namespace/sandbox runtimes)
    async fn ping_health_tcp(&self, port: u16, endpoint: &str) -> Result<()> {
        use tokio::io::{AsyncReadExt, AsyncWriteExt};
        use tokio::net::TcpStream;

        let addr = format!("127.0.0.1:{}", port);
        let mut stream = tokio::time::timeout(HEALTH_CHECK_TIMEOUT, TcpStream::connect(&addr))
            .await
            .context("TCP connection timeout")?
            .context("Failed to connect")?;

        let request = format!(
            "GET {} HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n",
            endpoint
        );
        stream
            .write_all(request.as_bytes())
            .await
            .context("Failed to write request")?;

        let mut response = vec![0u8; 1024];
        let n = tokio::time::timeout(HEALTH_CHECK_TIMEOUT, stream.read(&mut response))
            .await
            .context("Read timeout")?
            .context("Failed to read response")?;

        let response_str = String::from_utf8_lossy(&response[..n]);
        if response_str.contains("200") {
            Ok(())
        } else {
            anyhow::bail!("Unhealthy response")
        }
    }

    /// Ping a health endpoint, optionally using vsock CONNECT protocol
    ///
    /// For Firecracker VMs, the vsock socket requires the CONNECT protocol:
    /// 1. Connect to the Unix socket (exposed by Firecracker)
    /// 2. Send "CONNECT <port>\n"
    /// 3. Wait for "OK <port>\n" response
    /// 4. Socket is now connected to guest app
    async fn ping_health_with_vsock(
        &self,
        socket_path: &PathBuf,
        endpoint: &str,
        vsock_port: Option<u32>,
    ) -> Result<()> {
        use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader};
        use tokio::net::UnixStream;

        let stream = tokio::time::timeout(HEALTH_CHECK_TIMEOUT, UnixStream::connect(socket_path))
            .await
            .context("Connection timeout")?
            .context("Failed to connect")?;

        let (reader, mut writer) = stream.into_split();
        let mut reader = BufReader::new(reader);

        // If vsock port is specified, perform CONNECT handshake
        if let Some(port) = vsock_port {
            // Send CONNECT request
            let connect_cmd = format!("CONNECT {}\n", port);
            writer
                .write_all(connect_cmd.as_bytes())
                .await
                .context("Failed to send CONNECT")?;

            // Read response line
            let mut response_line = String::new();
            tokio::time::timeout(HEALTH_CHECK_TIMEOUT, reader.read_line(&mut response_line))
                .await
                .context("CONNECT response timeout")?
                .context("Failed to read CONNECT response")?;

            // Expect "OK <port>\n"
            let expected = format!("OK {}", port);
            if !response_line.starts_with(&expected) {
                anyhow::bail!(
                    "VSOCK CONNECT failed: expected '{}', got '{}'",
                    expected,
                    response_line.trim()
                );
            }
        }

        // Now send HTTP health check request
        let request = format!(
            "GET {} HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n",
            endpoint
        );
        writer
            .write_all(request.as_bytes())
            .await
            .context("Failed to write request")?;

        let mut response = vec![0u8; 1024];
        let n = tokio::time::timeout(HEALTH_CHECK_TIMEOUT, reader.read(&mut response))
            .await
            .context("Read timeout")?
            .context("Failed to read response")?;

        let response_str = String::from_utf8_lossy(&response[..n]);
        if response_str.contains("200 OK") {
            Ok(())
        } else {
            anyhow::bail!("Unhealthy response")
        }
    }

    /// Run health checks on all instances and handle unhealthy ones
    pub async fn run_health_checks(&self) {
        let instance_ids: Vec<InstanceId> = {
            let instances = self.instances.read().await;
            instances.keys().cloned().collect()
        };

        for instance_id in instance_ids {
            let status = self.check_health(&instance_id.process, &instance_id.id).await;

            match status {
                HealthStatus::Unhealthy => {
                    info!("Instance {} is unhealthy, restarting", instance_id);
                    if let Err(e) = self.restart(&instance_id.process, &instance_id.id).await {
                        error!("Failed to restart {}: {}", instance_id, e);
                    }
                }
                HealthStatus::Failed => {
                    error!("Instance {} has failed (too many restarts)", instance_id);
                }
                _ => {}
            }
        }
    }

    /// Start the background health monitor loop
    pub fn start_monitor(self: Arc<Self>) {
        let interval = Duration::from_secs(self.config.settings.health_check_interval);
        let hyp = self.clone();
        tokio::spawn(async move {
            info!("Starting health monitor (interval: {:?})", interval);
            loop {
                tokio::time::sleep(interval).await;
                hyp.run_health_checks().await;
                hyp.reap_idle_instances().await;
                hyp.check_storage_quotas().await;
            }
        });
    }

    /// Update activity timestamp for an instance.
    /// Call this on real requests (NOT health checks) to prevent auto-stop.
    pub async fn touch_activity(&self, process_name: &str, id: &str) {
        let instance_id = InstanceId::new(process_name, id);
        let mut instances = self.instances.write().await;
        if let Some(instance) = instances.get_mut(&instance_id) {
            instance.touch();
        }
    }

    /// Set the traffic weight for an instance (0-100).
    /// Weight 0 means the instance receives no traffic.
    /// Weight 100 is the default and means full traffic.
    /// Returns Err if the instance is not found.
    pub async fn set_weight(&self, process_name: &str, id: &str, weight: u8) -> Result<()> {
        let instance_id = InstanceId::new(process_name, id);
        let mut instances = self.instances.write().await;
        if let Some(instance) = instances.get_mut(&instance_id) {
            instance.weight = weight.min(100); // Cap at 100
            info!("Set weight for {} to {}", instance_id, instance.weight);
            Ok(())
        } else {
            anyhow::bail!("Instance not found: {}", instance_id)
        }
    }

    /// List all running instances for a specific process.
    /// Used for weighted load balancing across multiple instances.
    pub async fn list_by_process(&self, process_name: &str) -> Vec<InstanceInfo> {
        let instances = self.instances.read().await;
        instances
            .values()
            .filter(|i| i.id.process == process_name)
            .map(|i| i.info())
            .collect()
    }

    /// Select an instance for a process using weighted random selection.
    /// Returns None if no instances are available or all have weight 0.
    pub async fn select_weighted(&self, process_name: &str) -> Option<InstanceInfo> {
        use rand::Rng;

        let instances = self.instances.read().await;
        let candidates: Vec<_> = instances
            .values()
            .filter(|i| i.id.process == process_name && i.weight > 0)
            .collect();

        if candidates.is_empty() {
            return None;
        }

        // Calculate total weight
        let total_weight: u32 = candidates.iter().map(|i| i.weight as u32).sum();
        if total_weight == 0 {
            return None;
        }

        // Pick a random point in the weight space
        let mut rng = rand::thread_rng();
        let point = rng.gen_range(0..total_weight);

        // Find the instance at that point
        let mut cumulative = 0u32;
        for instance in candidates {
            cumulative += instance.weight as u32;
            if point < cumulative {
                return Some(instance.info());
            }
        }

        // Fallback (shouldn't happen)
        None
    }

    /// Stop idle instances that have exceeded their idle_timeout.
    /// Called periodically by the health monitor.
    async fn reap_idle_instances(&self) {
        let idle_instances: Vec<InstanceId> = {
            let instances = self.instances.read().await;
            instances
                .values()
                .filter(|i| i.is_idle())
                .map(|i| i.id.clone())
                .collect()
        };

        for instance_id in idle_instances {
            // Don't reap instances with active connections
            let active = self
                .active_connection_count(&instance_id.process, &instance_id.id)
                .await;
            if active > 0 {
                info!(
                    "Instance {} is idle but has {} active connection(s), skipping reap",
                    instance_id, active
                );
                continue;
            }

            let idle_secs = {
                let instances = self.instances.read().await;
                instances
                    .get(&instance_id)
                    .map(|i| i.last_activity.elapsed().as_secs())
                    .unwrap_or(0)
            };

            info!(
                "Stopping idle instance {} (idle: {}s)",
                instance_id, idle_secs
            );

            if let Err(e) = self.stop(&instance_id.process, &instance_id.id).await {
                error!("Failed to stop idle instance {}: {}", instance_id, e);
            }
        }
    }

    /// Check storage quotas for all instances and update metrics.
    /// Logs warnings at 80% and errors at 100% usage.
    async fn check_storage_quotas(&self) {
        let instance_data: Vec<(InstanceId, std::path::PathBuf, Option<u32>)> = {
            let instances = self.instances.read().await;
            instances
                .values()
                .map(|i| (i.id.clone(), i.data_dir.clone(), i.storage_quota_mb))
                .collect()
        };

        for (instance_id, data_dir, quota_mb) in instance_data {
            let used_bytes = match calculate_dir_size(data_dir.clone()).await {
                Ok(bytes) => bytes,
                Err(e) => {
                    error!("Failed to calculate storage for {}: {}", instance_id, e);
                    continue;
                }
            };

            // Update cached storage usage on the instance
            {
                let mut instances = self.instances.write().await;
                if let Some(instance) = instances.get_mut(&instance_id) {
                    instance.storage_used_bytes = used_bytes;
                }
            }

            // Update metrics
            let mut labels = HashMap::new();
            labels.insert("process".to_string(), instance_id.process.clone());
            labels.insert("id".to_string(), instance_id.id.clone());
            let gauge = self.metrics.instance_storage_bytes.with_labels(&labels).await;
            gauge.set(used_bytes);

            // Check quota
            if let Some(quota_mb) = quota_mb {
                let quota_bytes = (quota_mb as u64) * 1024 * 1024;

                let quota_gauge = self.metrics.instance_storage_quota_bytes.with_labels(&labels).await;
                quota_gauge.set(quota_bytes);

                if quota_bytes > 0 {
                    let ratio = used_bytes as f64 / quota_bytes as f64;
                    let ratio_gauge = self.metrics.instance_storage_usage_ratio.with_labels(&labels).await;
                    ratio_gauge.set((ratio * 10000.0) as u64);

                    if ratio >= 1.0 {
                        error!(
                            "Instance {} storage quota exceeded: {:.1}MB / {:.1}MB ({:.0}%)",
                            instance_id,
                            used_bytes as f64 / 1024.0 / 1024.0,
                            quota_mb as f64,
                            ratio * 100.0
                        );
                    } else if ratio >= 0.8 {
                        warn!(
                            "Instance {} storage usage high: {:.1}MB / {:.1}MB ({:.0}%)",
                            instance_id,
                            used_bytes as f64 / 1024.0 / 1024.0,
                            quota_mb as f64,
                            ratio * 100.0
                        );
                    }
                }
            }
        }
    }

    /// Recover orphaned instances from a previous crash.
    /// Checks persisted state, kills any still-running orphans, and cleans up.
    /// Called on startup before spawning configured instances.
    pub async fn recover_orphans(&self) {
        let store = match &self.state_store {
            Some(s) => s,
            None => return,
        };

        let states = match store.list().await {
            Ok(s) => s,
            Err(e) => {
                error!("Failed to read instance state for recovery: {}", e);
                return;
            }
        };

        if states.is_empty() {
            return;
        }

        info!("Found {} orphaned instance(s) from previous run", states.len());

        for state in &states {
            // Check if process is still alive
            #[cfg(unix)]
            let alive = unsafe { libc::kill(state.pid as i32, 0) } == 0;
            #[cfg(not(unix))]
            let alive = false;

            if alive {
                info!(
                    "Killing orphaned process {} (pid {})",
                    state.instance_id, state.pid
                );
                #[cfg(unix)]
                unsafe {
                    libc::kill(state.pid as i32, libc::SIGKILL);
                }
            } else {
                info!(
                    "Orphaned process {} (pid {}) already exited",
                    state.instance_id, state.pid
                );
            }

            // Note: port allocator is in-memory and starts fresh on each run,
            // so no port release needed during recovery.
        }

        // Clear all persisted state
        if let Err(e) = store.clear_all().await {
            error!("Failed to clear instance state after recovery: {}", e);
        }

        info!("Orphan recovery complete");
    }

    /// Spawn all instances configured in [instances] section.
    /// Called on server startup to auto-start configured instances.
    /// Continues spawning even if some fail, logs errors for failures.
    /// Returns the number of successfully spawned instances.
    pub async fn spawn_configured_instances(&self) -> (usize, usize) {
        let instances_to_spawn = self.config.get_instances_to_spawn();

        if instances_to_spawn.is_empty() {
            return (0, 0);
        }

        info!("Auto-spawning {} configured instance(s)", instances_to_spawn.len());

        let mut success_count = 0;
        let mut fail_count = 0;

        for (service_name, instance_id) in instances_to_spawn {
            info!("Auto-spawning {}:{}", service_name, instance_id);

            match self.spawn(&service_name, &instance_id).await {
                Ok(socket) => {
                    info!(
                        "Successfully spawned {}:{} at {:?}",
                        service_name, instance_id, socket
                    );
                    success_count += 1;
                }
                Err(e) => {
                    error!(
                        "Failed to spawn {}:{}: {}",
                        service_name, instance_id, e
                    );
                    fail_count += 1;
                }
            }
        }

        if fail_count > 0 {
            warn!(
                "Auto-spawn complete: {} succeeded, {} failed",
                success_count, fail_count
            );
        } else {
            info!("Auto-spawn complete: {} instance(s) started", success_count);
        }

        (success_count, fail_count)
    }

    /// Spawn instance if not running, and wait for it to be ready.
    /// Returns the socket path. Use this for wake-on-request.
    /// Uses the process's configured startup_timeout (default: 10s).
    pub async fn spawn_and_wait(&self, process_name: &str, id: &str) -> Result<PathBuf> {
        let instance_id = InstanceId::new(process_name, id);

        // Wake-once pattern: if another request is already waking this instance,
        // wait for it to finish instead of spawning a duplicate.
        // Wake-once: if another request is already waking this instance, wait for it.
        let maybe_notify = {
            let waking = self.waking.read().await;
            waking.get(&instance_id).cloned()
        };
        if let Some(notify) = maybe_notify {
            // Register for notification BEFORE checking if spawn is done.
            // This avoids missing the signal if it fires between our check and await.
            let notified = notify.notified();
            info!("Instance {} is already being woken, waiting", instance_id);
            // Check if instance became available while we were setting up
            if self.is_running(process_name, id).await {
                self.touch_activity(process_name, id).await;
                let info = self.get(process_name, id).await
                    .ok_or_else(|| anyhow::anyhow!("Instance {} disappeared", instance_id))?;
                return Ok(info.socket);
            }
            notified.await;
            if let Some(info) = self.get(process_name, id).await {
                self.touch_activity(process_name, id).await;
                return Ok(info.socket);
            }
            anyhow::bail!("Instance {} failed to wake", instance_id);
        }

        // We're the first to wake this instance. Register a Notify.
        let notify = Arc::new(tokio::sync::Notify::new());
        {
            let mut waking = self.waking.write().await;
            waking.insert(instance_id.clone(), notify.clone());
        }

        // Get the startup timeout from process config
        let timeout_secs = self
            .config
            .get_service(process_name)
            .map(|p| p.startup_timeout)
            .unwrap_or(10);

        // spawn_if_not_running already waits for TCP/socket readiness via spawn()
        let socket = self.spawn_if_not_running(process_name, id).await?;

        // Get port info to determine readiness check method
        let port = self
            .get(process_name, id)
            .await
            .and_then(|info| info.port);

        // Wait for service to be ready (check every 100ms)
        // Try TCP first if port is available, fall back to socket existence
        let iterations = (timeout_secs * 10) as usize;
        let mut ready = false;
        for _ in 0..iterations {
            let is_ready = if let Some(port) = port {
                // TCP mode: try to connect
                if tokio::net::TcpStream::connect(format!("127.0.0.1:{}", port))
                    .await
                    .is_ok()
                {
                    true
                } else {
                    // Fall back to socket existence check (test stubs may not listen on TCP)
                    socket.exists()
                }
            } else {
                socket.exists()
            };

            if is_ready {
                self.touch_activity(process_name, id).await;
                ready = true;
                break;
            }
            tokio::time::sleep(Duration::from_millis(100)).await;
        }

        // Notify all waiters and remove the Notify
        notify.notify_waiters();
        {
            let mut waking = self.waking.write().await;
            waking.remove(&instance_id);
        }

        if ready {
            Ok(socket)
        } else {
            anyhow::bail!("Instance {} failed to start within {} seconds",
                instance_id, timeout_secs)
        }
    }

    /// Deploy a new instance version and wait for it to be healthy.
    /// Used for blue/green and canary deployments.
    ///
    /// - Spawns the instance with the given version as the instance ID
    /// - Sets the initial weight (default 100)
    /// - Waits for health check to pass (timeout ~30s)
    /// - Returns error if health check fails within timeout
    pub async fn deploy_and_wait_healthy(
        &self,
        process_name: &str,
        version: &str,
        initial_weight: u8,
        timeout_secs: u64,
    ) -> Result<PathBuf> {
        let instance_id = InstanceId::new(process_name, version);

        // Spawn the instance
        let socket = self.spawn(process_name, version).await?;

        // Set initial weight
        self.set_weight(process_name, version, initial_weight).await?;

        // Wait for health check to pass
        let check_interval = Duration::from_millis(500);
        let timeout = Duration::from_secs(timeout_secs);
        let start = Instant::now();

        while start.elapsed() < timeout {
            let status = self.check_health(process_name, version).await;
            match status {
                HealthStatus::Healthy => {
                    info!("Instance {} is healthy", instance_id);
                    return Ok(socket);
                }
                HealthStatus::Failed => {
                    anyhow::bail!("Instance {} health check failed permanently", instance_id);
                }
                _ => {
                    // Unknown, Degraded, or Unhealthy - keep waiting
                    tokio::time::sleep(check_interval).await;
                }
            }
        }

        // Timeout reached - stop the unhealthy instance and return error
        let _ = self.stop(process_name, version).await;
        anyhow::bail!(
            "Instance {} did not become healthy within {} seconds",
            instance_id,
            timeout_secs
        )
    }

    /// Atomically swap traffic weights between two versions.
    /// Sets `from_version` weight to 0 and `to_version` weight to 100.
    /// Used for blue/green instant cutover.
    ///
    /// Returns error if either version is not running.
    pub async fn route_swap(
        &self,
        process_name: &str,
        from_version: &str,
        to_version: &str,
    ) -> Result<()> {
        let from_id = InstanceId::new(process_name, from_version);
        let to_id = InstanceId::new(process_name, to_version);

        // Acquire write lock and update both weights atomically
        let mut instances = self.instances.write().await;

        // Verify both instances exist
        if !instances.contains_key(&from_id) {
            anyhow::bail!("Instance {} not found", from_id);
        }
        if !instances.contains_key(&to_id) {
            anyhow::bail!("Instance {} not found", to_id);
        }

        // Atomically update weights
        if let Some(from_instance) = instances.get_mut(&from_id) {
            from_instance.weight = 0;
        }
        if let Some(to_instance) = instances.get_mut(&to_id) {
            to_instance.weight = 100;
        }

        info!(
            "Traffic swap complete: {} weight=0, {} weight=100",
            from_id, to_id
        );

        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::config::ProcessConfig;
    use std::collections::HashMap;
    use std::path::Path;
    use tempfile::TempDir;

    // Helper to create a test config with a simple echo server process
    fn test_config_with_process(name: &str, command: &str, args: Vec<&str>) -> Config {
        let test_id = format!("{}-{}", std::process::id(), rand::random::<u32>());
        let mut config = Config::default();
        config.settings.data_dir = std::env::temp_dir().join(format!("tenement-test-{}", test_id));

        let process = ProcessConfig {
            command: command.to_string(),
            args: args.into_iter().map(|s| s.to_string()).collect(),
            socket: format!("/tmp/tenement-test-{}/{{name}}-{{id}}.sock", test_id),
            isolation: RuntimeType::Process,
            health: None,
            env: HashMap::new(),
            workdir: None,
            restart: "on-failure".to_string(),
            idle_timeout: None,
            startup_timeout: 5,
            request_timeout: 30,
            memory_limit_mb: None,
            cpu_shares: None,
            kernel: None,
            rootfs: None,
            memory_mb: 256,
            vcpus: 1,
            vsock_port: 5000,
            storage_quota_mb: None,
            storage_persist: false,
        };

        config.service.insert(name.to_string(), process);
        config
    }

    // Helper to create a shell script that creates a socket and waits
    fn create_socket_server_script(dir: &Path) -> PathBuf {
        let script_path = dir.join("server.sh");
        let script = r#"#!/bin/bash
SOCKET_PATH="${SOCKET_PATH:-/tmp/test.sock}"
rm -f "$SOCKET_PATH"
# Create socket using nc (netcat) - listens and responds
# Use socat if available, otherwise use a simple approach
if command -v socat &> /dev/null; then
    socat UNIX-LISTEN:"$SOCKET_PATH",fork EXEC:"echo HTTP/1.1 200 OK"
elif command -v nc &> /dev/null; then
    while true; do
        echo -e "HTTP/1.1 200 OK\r\n\r\nOK" | nc -lU "$SOCKET_PATH" -q0 2>/dev/null || break
    done
else
    # Fallback: just create the socket file and sleep
    python3 -c "
import socket
import os
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
sock.bind('$SOCKET_PATH')
sock.listen(1)
while True:
    conn, _ = sock.accept()
    conn.sendall(b'HTTP/1.1 200 OK\r\n\r\nOK')
    conn.close()
" 2>/dev/null &
    sleep infinity
fi
"#;
        std::fs::write(&script_path, script).unwrap();
        #[cfg(unix)]
        {
            use std::os::unix::fs::PermissionsExt;
            std::fs::set_permissions(&script_path, std::fs::Permissions::from_mode(0o755)).unwrap();
        }
        script_path
    }

    // Helper to create a simple process that touches a socket file
    fn create_touch_socket_script(dir: &Path) -> PathBuf {
        let script_path = dir.join("touch_socket.sh");
        let script = r#"#!/bin/bash
SOCKET_PATH="${SOCKET_PATH:-/tmp/test.sock}"
rm -f "$SOCKET_PATH"
touch "$SOCKET_PATH"
sleep 30
"#;
        std::fs::write(&script_path, script).unwrap();
        #[cfg(unix)]
        {
            use std::os::unix::fs::PermissionsExt;
            std::fs::set_permissions(&script_path, std::fs::Permissions::from_mode(0o755)).unwrap();
        }
        script_path
    }

    // ===================
    // BACKOFF TESTS
    // ===================

    #[test]
    fn test_calculate_backoff() {
        let config = Config::default();
        let hypervisor = Hypervisor::new(config);

        // Default settings: base=1000ms, max=60000ms

        // 0 restarts = no delay
        assert_eq!(hypervisor.calculate_backoff(0), Duration::ZERO);

        // 1 restart = 1000ms (1s)
        assert_eq!(hypervisor.calculate_backoff(1), Duration::from_millis(1000));

        // 2 restarts = 2000ms (2s)
        assert_eq!(hypervisor.calculate_backoff(2), Duration::from_millis(2000));

        // 3 restarts = 4000ms (4s)
        assert_eq!(hypervisor.calculate_backoff(3), Duration::from_millis(4000));

        // 4 restarts = 8000ms (8s)
        assert_eq!(hypervisor.calculate_backoff(4), Duration::from_millis(8000));

        // 5 restarts = 16000ms (16s)
        assert_eq!(hypervisor.calculate_backoff(5), Duration::from_millis(16000));

        // 6 restarts = 32000ms (32s)
        assert_eq!(hypervisor.calculate_backoff(6), Duration::from_millis(32000));

        // 7 restarts = 64000ms but capped at 60000ms (60s max)
        assert_eq!(hypervisor.calculate_backoff(7), Duration::from_millis(60000));

        // Large values stay capped
        assert_eq!(hypervisor.calculate_backoff(100), Duration::from_millis(60000));
    }

    #[test]
    fn test_calculate_backoff_custom_settings() {
        let mut config = Config::default();
        config.settings.backoff_base_ms = 500;
        config.settings.backoff_max_ms = 5000;
        let hypervisor = Hypervisor::new(config);

        // 1 restart = 500ms
        assert_eq!(hypervisor.calculate_backoff(1), Duration::from_millis(500));

        // 2 restarts = 1000ms
        assert_eq!(hypervisor.calculate_backoff(2), Duration::from_millis(1000));

        // 4 restarts = 4000ms
        assert_eq!(hypervisor.calculate_backoff(4), Duration::from_millis(4000));

        // 5 restarts = 8000ms but capped at 5000ms
        assert_eq!(hypervisor.calculate_backoff(5), Duration::from_millis(5000));
    }

    #[test]
    fn test_calculate_backoff_overflow_protection() {
        let mut config = Config::default();
        config.settings.backoff_base_ms = u64::MAX;
        config.settings.backoff_max_ms = u64::MAX;
        let hypervisor = Hypervisor::new(config);

        // Should not panic or overflow, should saturate
        let result = hypervisor.calculate_backoff(100);
        assert_eq!(result, Duration::from_millis(u64::MAX));
    }

    #[test]
    fn test_calculate_backoff_zero_base() {
        let mut config = Config::default();
        config.settings.backoff_base_ms = 0;
        config.settings.backoff_max_ms = 1000;
        let hypervisor = Hypervisor::new(config);

        // 0 * 2^n = 0
        assert_eq!(hypervisor.calculate_backoff(1), Duration::ZERO);
        assert_eq!(hypervisor.calculate_backoff(5), Duration::ZERO);
    }

    // ===================
    // LIFECYCLE TESTS
    // ===================

    #[tokio::test]
    async fn test_spawn_process_instance() {
        let dir = TempDir::new().unwrap();
        let script = create_touch_socket_script(dir.path());

        let config = test_config_with_process("test-api", script.to_str().unwrap(), vec![]);
        let hypervisor = Hypervisor::new(config);

        let result = hypervisor.spawn("test-api", "instance1").await;
        assert!(result.is_ok(), "Spawn should succeed: {:?}", result.err());

        // Verify instance is tracked
        assert!(hypervisor.is_running("test-api", "instance1").await);

        // Clean up
        hypervisor.stop("test-api", "instance1").await.ok();
    }

    #[tokio::test]
    async fn test_spawn_creates_instance_entry() {
        let dir = TempDir::new().unwrap();
        let script = create_touch_socket_script(dir.path());

        let config = test_config_with_process("api", script.to_str().unwrap(), vec![]);
        let hypervisor = Hypervisor::new(config);

        // Initially empty
        let list = hypervisor.list().await;
        assert!(list.is_empty());

        // Spawn
        hypervisor.spawn("api", "user1").await.unwrap();

        // Should appear in list
        let list = hypervisor.list().await;
        assert_eq!(list.len(), 1);
        assert_eq!(list[0].id.process, "api");
        assert_eq!(list[0].id.id, "user1");

        // Clean up
        hypervisor.stop("api", "user1").await.ok();
    }

    #[tokio::test]
    async fn test_spawn_returns_socket_path() {
        let dir = TempDir::new().unwrap();
        let script = create_touch_socket_script(dir.path());

        let config = test_config_with_process("myapp", script.to_str().unwrap(), vec![]);
        let hypervisor = Hypervisor::new(config);

        let socket = hypervisor.spawn("myapp", "prod").await.unwrap();
        assert!(socket.to_string_lossy().contains("myapp"));
        assert!(socket.to_string_lossy().contains("prod"));

        hypervisor.stop("myapp", "prod").await.ok();
    }

    #[tokio::test]
    async fn test_spawn_with_command_string_shell_splits() {
        // When command is "echo hello world" with no explicit args,
        // it should be shell-split into command="echo", args=["hello", "world"]
        let config = test_config_with_process("api", "echo hello world", vec![]);
        let hypervisor = Hypervisor::new(config);

        let result = hypervisor.spawn("api", "test").await;
        assert!(result.is_ok(), "Shell-split command should spawn: {:?}", result.err());

        hypervisor.stop("api", "test").await.ok();
    }

    #[tokio::test]
    async fn test_spawn_with_explicit_args_no_split() {
        // When args are explicitly provided, command should NOT be split
        let config = test_config_with_process("api", "echo", vec!["hello", "world"]);
        let hypervisor = Hypervisor::new(config);

        let result = hypervisor.spawn("api", "test").await;
        assert!(result.is_ok(), "Explicit args should spawn: {:?}", result.err());

        hypervisor.stop("api", "test").await.ok();
    }

    #[tokio::test]
    async fn test_spawn_with_interpolation_and_shell_split() {
        // Command with interpolation vars should be interpolated first, then shell-split
        let config = test_config_with_process("api", "echo --id {id} --name {name}", vec![]);
        let hypervisor = Hypervisor::new(config);

        // Should interpolate to "echo --id user123 --name api" then split into
        // command="echo", args=["--id", "user123", "--name", "api"]
        let result = hypervisor.spawn("api", "user123").await;
        assert!(result.is_ok(), "Interpolated+split command should spawn: {:?}", result.err());

        hypervisor.stop("api", "user123").await.ok();
    }

    #[tokio::test]
    async fn test_stop_instance() {
        let dir = TempDir::new().unwrap();
        let script = create_touch_socket_script(dir.path());

        let config = test_config_with_process("api", script.to_str().unwrap(), vec![]);
        let hypervisor = Hypervisor::new(config);

        hypervisor.spawn("api", "test").await.unwrap();
        assert!(hypervisor.is_running("api", "test").await);

        let result = hypervisor.stop("api", "test").await;
        assert!(result.is_ok());
        assert!(!hypervisor.is_running("api", "test").await);
    }

    #[tokio::test]
    async fn test_stop_nonexistent_instance_returns_error() {
        let config = Config::default();
        let hypervisor = Hypervisor::new(config);

        let result = hypervisor.stop("nonexistent", "id").await;
        assert!(result.is_err());
        assert!(result.unwrap_err().to_string().contains("not found"));
    }

    #[tokio::test]
    async fn test_list_instances() {
        let dir = TempDir::new().unwrap();
        let script = create_touch_socket_script(dir.path());

        let config = test_config_with_process("api", script.to_str().unwrap(), vec![]);
        let hypervisor = Hypervisor::new(config);

        // Spawn multiple instances
        hypervisor.spawn("api", "user1").await.unwrap();
        hypervisor.spawn("api", "user2").await.unwrap();

        let list = hypervisor.list().await;
        assert_eq!(list.len(), 2);

        // Verify both are present
        let ids: Vec<String> = list.iter().map(|i| i.id.id.clone()).collect();
        assert!(ids.contains(&"user1".to_string()));
        assert!(ids.contains(&"user2".to_string()));

        // Clean up
        hypervisor.stop("api", "user1").await.ok();
        hypervisor.stop("api", "user2").await.ok();
    }

    #[tokio::test]
    async fn test_get_instance() {
        let dir = TempDir::new().unwrap();
        let script = create_touch_socket_script(dir.path());

        let config = test_config_with_process("api", script.to_str().unwrap(), vec![]);
        let hypervisor = Hypervisor::new(config);

        hypervisor.spawn("api", "user1").await.unwrap();

        let info = hypervisor.get("api", "user1").await;
        assert!(info.is_some());
        let info = info.unwrap();
        assert_eq!(info.id.process, "api");
        assert_eq!(info.id.id, "user1");

        // Non-existent should return None
        assert!(hypervisor.get("api", "nonexistent").await.is_none());

        hypervisor.stop("api", "user1").await.ok();
    }

    #[tokio::test]
    async fn test_spawn_if_not_running() {
        let dir = TempDir::new().unwrap();
        let script = create_touch_socket_script(dir.path());

        let config = test_config_with_process("api", script.to_str().unwrap(), vec![]);
        let hypervisor = Hypervisor::new(config);

        // First call spawns
        let socket1 = hypervisor.spawn_if_not_running("api", "test").await.unwrap();

        // Second call returns existing socket without spawning again
        let socket2 = hypervisor.spawn_if_not_running("api", "test").await.unwrap();
        assert_eq!(socket1, socket2);

        // Only one instance in list
        let list = hypervisor.list().await;
        assert_eq!(list.len(), 1);

        hypervisor.stop("api", "test").await.ok();
    }

    #[tokio::test]
    async fn test_spawn_already_running_returns_socket() {
        let dir = TempDir::new().unwrap();
        let script = create_touch_socket_script(dir.path());

        let config = test_config_with_process("api", script.to_str().unwrap(), vec![]);
        let hypervisor = Hypervisor::new(config);

        let socket1 = hypervisor.spawn("api", "test").await.unwrap();
        let socket2 = hypervisor.spawn("api", "test").await.unwrap();

        // Both return same socket (idempotent)
        assert_eq!(socket1, socket2);

        hypervisor.stop("api", "test").await.ok();
    }

    // ===================
    // ERROR PATH TESTS
    // ===================

    #[tokio::test]
    async fn test_spawn_unknown_process_returns_error() {
        let config = Config::default();
        let hypervisor = Hypervisor::new(config);

        let result = hypervisor.spawn("nonexistent", "id").await;
        assert!(result.is_err());
        assert!(result.unwrap_err().to_string().contains("Unknown process"));
    }

    #[tokio::test]
    async fn test_spawn_command_not_found() {
        let config = test_config_with_process("api", "/nonexistent/binary", vec![]);
        let hypervisor = Hypervisor::new(config);

        let result = hypervisor.spawn("api", "test").await;
        assert!(result.is_err());
    }

    #[tokio::test]
    async fn test_has_process() {
        let config = test_config_with_process("myapi", "sleep", vec!["1"]);
        let hypervisor = Hypervisor::new(config);

        assert!(hypervisor.has_process("myapi"));
        assert!(!hypervisor.has_process("unknown"));
    }

    // ===================
    // RESTART TESTS
    // ===================

    #[tokio::test]
    async fn test_restart_instance() {
        let dir = TempDir::new().unwrap();
        let script = create_touch_socket_script(dir.path());

        let mut config = test_config_with_process("api", script.to_str().unwrap(), vec![]);
        // Use zero backoff for faster tests
        config.settings.backoff_base_ms = 0;
        let hypervisor = Hypervisor::new(config);

        hypervisor.spawn("api", "test").await.unwrap();

        // Restart
        let result = hypervisor.restart("api", "test").await;
        assert!(result.is_ok());

        // Should still be running
        assert!(hypervisor.is_running("api", "test").await);

        hypervisor.stop("api", "test").await.ok();
    }

    #[tokio::test]
    async fn test_restart_increments_counter() {
        let dir = TempDir::new().unwrap();
        let script = create_touch_socket_script(dir.path());

        let mut config = test_config_with_process("api", script.to_str().unwrap(), vec![]);
        config.settings.backoff_base_ms = 0;
        let hypervisor = Hypervisor::new(config);

        hypervisor.spawn("api", "test").await.unwrap();

        // Check initial restart count
        let info = hypervisor.get("api", "test").await.unwrap();
        assert_eq!(info.restarts, 0);

        // Restart
        hypervisor.restart("api", "test").await.unwrap();

        // Restart count should be 1
        let info = hypervisor.get("api", "test").await.unwrap();
        assert_eq!(info.restarts, 1);

        hypervisor.stop("api", "test").await.ok();
    }

    // ===================
    // ACTIVITY TRACKING TESTS
    // ===================

    #[tokio::test]
    async fn test_touch_activity() {
        let dir = TempDir::new().unwrap();
        let script = create_touch_socket_script(dir.path());

        let config = test_config_with_process("api", script.to_str().unwrap(), vec![]);
        let hypervisor = Hypervisor::new(config);

        hypervisor.spawn("api", "test").await.unwrap();

        // Wait a bit
        tokio::time::sleep(Duration::from_millis(50)).await;

        // Get idle time before touch
        let info_before = hypervisor.get("api", "test").await.unwrap();
        let idle_before = info_before.idle_secs;

        // Touch activity
        hypervisor.touch_activity("api", "test").await;

        // Idle time should reset (or be very small)
        let info_after = hypervisor.get("api", "test").await.unwrap();
        assert!(info_after.idle_secs <= idle_before);

        hypervisor.stop("api", "test").await.ok();
    }

    #[tokio::test]
    async fn test_get_and_touch_running_instance() {
        let dir = TempDir::new().unwrap();
        let script = create_touch_socket_script(dir.path());

        let config = test_config_with_process("api", script.to_str().unwrap(), vec![]);
        let hypervisor = Hypervisor::new(config);

        hypervisor.spawn("api", "test").await.unwrap();

        // Wait a bit so idle time accumulates
        tokio::time::sleep(Duration::from_millis(50)).await;

        // get_and_touch should return Some and reset activity
        let info = hypervisor.get_and_touch("api", "test").await;
        assert!(info.is_some());
        let info = info.unwrap();
        assert_eq!(info.id.process, "api");
        assert_eq!(info.id.id, "test");

        // Idle time should be reset (very small after touch)
        assert!(info.idle_secs < 1);

        hypervisor.stop("api", "test").await.ok();
    }

    #[tokio::test]
    async fn test_get_and_touch_nonexistent_instance() {
        let config = Config::default();
        let hypervisor = Hypervisor::new(config);

        // Non-existent instance should return None
        let info = hypervisor.get_and_touch("api", "nonexistent").await;
        assert!(info.is_none());
    }

    #[tokio::test]
    async fn test_get_and_touch_is_atomic() {
        // This test verifies get_and_touch provides instance info AND touches activity
        // in a single operation (vs separate is_running + touch_activity + get calls)
        let dir = TempDir::new().unwrap();
        let script = create_touch_socket_script(dir.path());

        let config = test_config_with_process("api", script.to_str().unwrap(), vec![]);
        let hypervisor = Hypervisor::new(config);

        hypervisor.spawn("api", "test").await.unwrap();

        // Wait so idle time > 0
        tokio::time::sleep(Duration::from_millis(100)).await;

        // Record time before
        let before_info = hypervisor.get("api", "test").await.unwrap();
        let idle_before = before_info.idle_secs;

        // Wait a bit more
        tokio::time::sleep(Duration::from_millis(50)).await;

        // get_and_touch should touch activity
        let touched_info = hypervisor.get_and_touch("api", "test").await.unwrap();

        // The returned info should have fresh activity timestamp
        // (idle_secs should be 0 or very small since we just touched)
        assert!(touched_info.idle_secs <= idle_before);

        // Verify activity was actually touched by checking again
        let after_info = hypervisor.get("api", "test").await.unwrap();
        assert!(after_info.idle_secs < 1); // Should be very fresh

        hypervisor.stop("api", "test").await.ok();
    }

    // ===================
    // LOG CAPTURE TESTS
    // ===================

    #[tokio::test]
    async fn test_spawn_captures_stdout() {
        let config = test_config_with_process("api", "echo", vec!["hello from stdout"]);
        let hypervisor = Hypervisor::new(config);

        hypervisor.spawn("api", "test").await.unwrap();

        // Give time for log capture
        tokio::time::sleep(Duration::from_millis(100)).await;

        // Check log buffer
        let logs = hypervisor.log_buffer().query(&crate::logs::LogQuery {
            process: Some("api".to_string()),
            ..Default::default()
        }).await;

        // Should have captured stdout
        assert!(logs.iter().any(|l| l.message.contains("hello from stdout")));
    }

    #[tokio::test]
    async fn test_spawn_captures_stderr() {
        let config = test_config_with_process("api", "sh", vec!["-c", "echo error >&2"]);
        let hypervisor = Hypervisor::new(config);

        hypervisor.spawn("api", "test").await.unwrap();

        // Give time for log capture
        tokio::time::sleep(Duration::from_millis(100)).await;

        // Check log buffer
        let logs = hypervisor.log_buffer().query(&crate::logs::LogQuery {
            process: Some("api".to_string()),
            level: Some(crate::logs::LogLevel::Stderr),
            ..Default::default()
        }).await;

        assert!(logs.iter().any(|l| l.message.contains("error")));
    }

    // ===================
    // METRICS TESTS
    // ===================

    #[tokio::test]
    async fn test_spawn_increments_metrics() {
        let dir = TempDir::new().unwrap();
        let script = create_touch_socket_script(dir.path());

        let config = test_config_with_process("api", script.to_str().unwrap(), vec![]);
        let hypervisor = Hypervisor::new(config);

        let metrics = hypervisor.metrics();
        let initial = metrics.instances_up.get();

        hypervisor.spawn("api", "test").await.unwrap();

        // instances_up should increment
        assert_eq!(metrics.instances_up.get(), initial + 1);

        hypervisor.stop("api", "test").await.unwrap();

        // Should decrement back
        assert_eq!(metrics.instances_up.get(), initial);
    }

    // ===================
    // HEALTH STATUS TESTS
    // ===================

    #[tokio::test]
    async fn test_check_health_no_endpoint_socket_file() {
        let dir = TempDir::new().unwrap();
        let script = create_touch_socket_script(dir.path());

        // No health endpoint configured, socket is just a file (not real socket)
        let config = test_config_with_process("api", script.to_str().unwrap(), vec![]);
        let hypervisor = Hypervisor::new(config);

        hypervisor.spawn("api", "test").await.unwrap();

        // Wait for socket file to be created
        tokio::time::sleep(Duration::from_millis(200)).await;

        let status = hypervisor.check_health("api", "test").await;
        // File exists but isn't a real socket, so can't connect
        // The actual status depends on implementation - could be Healthy (file exists)
        // or Unhealthy (can't connect). Just verify it returns a status.
        assert!(matches!(status, HealthStatus::Healthy | HealthStatus::Unhealthy));

        hypervisor.stop("api", "test").await.ok();
    }

    #[tokio::test]
    async fn test_check_health_unknown_process() {
        let config = Config::default();
        let hypervisor = Hypervisor::new(config);

        let status = hypervisor.check_health("nonexistent", "id").await;
        assert_eq!(status, HealthStatus::Unknown);
    }

    #[tokio::test]
    async fn test_check_health_not_running_instance() {
        let config = test_config_with_process("api", "sleep", vec!["1"]);
        let hypervisor = Hypervisor::new(config);

        // Clean up any stale socket from previous tests
        let socket_path = std::path::PathBuf::from("/tmp/api-test.sock");
        let _ = std::fs::remove_file(&socket_path);

        // Don't spawn, just check health
        // For a configured process that isn't running, returns Unhealthy
        let status = hypervisor.check_health("api", "test").await;
        assert!(matches!(status, HealthStatus::Unknown | HealthStatus::Unhealthy));
    }

    // ===================
    // DATA DIRECTORY TESTS
    // ===================

    #[tokio::test]
    async fn test_spawn_creates_data_directory() {
        let dir = TempDir::new().unwrap();
        let data_dir = dir.path().join("tenement-data");
        let script = create_touch_socket_script(dir.path());

        let mut config = test_config_with_process("api", script.to_str().unwrap(), vec![]);
        config.settings.data_dir = data_dir.clone();
        let hypervisor = Hypervisor::new(config);

        hypervisor.spawn("api", "user1").await.unwrap();

        // Data directory should be created
        assert!(data_dir.join("api").join("user1").exists());

        hypervisor.stop("api", "user1").await.ok();
    }

    // ===================
    // ENVIRONMENT VARIABLE TESTS
    // ===================

    #[tokio::test]
    async fn test_spawn_with_extra_env() {
        let config = test_config_with_process("api", "env", vec![]);
        let hypervisor = Hypervisor::new(config);

        let mut extra_env = HashMap::new();
        extra_env.insert("MY_CUSTOM_VAR".to_string(), "custom_value".to_string());

        hypervisor.spawn_with_env("api", "test", extra_env).await.unwrap();

        // Give time for process to run
        tokio::time::sleep(Duration::from_millis(100)).await;

        // Check that env was set (captured in logs)
        let logs = hypervisor.log_buffer().query(&crate::logs::LogQuery::default()).await;
        assert!(logs.iter().any(|l| l.message.contains("MY_CUSTOM_VAR=custom_value")));
    }

    #[tokio::test]
    async fn test_spawn_sets_port_env() {
        let config = test_config_with_process("api", "env", vec![]);
        let hypervisor = Hypervisor::new(config);

        hypervisor.spawn("api", "test").await.unwrap();

        tokio::time::sleep(Duration::from_millis(100)).await;

        // Process runtime should get a PORT env var with auto-allocated port (30000-40000)
        let logs = hypervisor.log_buffer().query(&crate::logs::LogQuery::default()).await;
        assert!(logs.iter().any(|l| l.message.contains("PORT=3")));
    }

    // ===================
    // AUTO-SPAWN TESTS
    // ===================

    #[tokio::test]
    async fn test_spawn_configured_instances_empty() {
        let config = Config::default();
        let hypervisor = Hypervisor::new(config);

        let (success, failed) = hypervisor.spawn_configured_instances().await;

        assert_eq!(success, 0);
        assert_eq!(failed, 0);
        assert!(hypervisor.list().await.is_empty());
    }

    #[tokio::test]
    async fn test_spawn_configured_instances_single() {
        let dir = TempDir::new().unwrap();
        let script = create_touch_socket_script(dir.path());

        let mut config = test_config_with_process("api", script.to_str().unwrap(), vec![]);
        config.instances.insert("api".to_string(), vec!["prod".to_string()]);
        let hypervisor = Hypervisor::new(config);

        let (success, failed) = hypervisor.spawn_configured_instances().await;

        assert_eq!(success, 1);
        assert_eq!(failed, 0);
        assert!(hypervisor.is_running("api", "prod").await);

        hypervisor.stop("api", "prod").await.ok();
    }

    #[tokio::test]
    async fn test_spawn_configured_instances_multiple() {
        let dir = TempDir::new().unwrap();
        let script = create_touch_socket_script(dir.path());

        let mut config = test_config_with_process("api", script.to_str().unwrap(), vec![]);
        config.instances.insert(
            "api".to_string(),
            vec!["prod".to_string(), "staging".to_string()],
        );
        let hypervisor = Hypervisor::new(config);

        let (success, failed) = hypervisor.spawn_configured_instances().await;

        assert_eq!(success, 2);
        assert_eq!(failed, 0);
        assert!(hypervisor.is_running("api", "prod").await);
        assert!(hypervisor.is_running("api", "staging").await);

        hypervisor.stop("api", "prod").await.ok();
        hypervisor.stop("api", "staging").await.ok();
    }

    #[tokio::test]
    async fn test_spawn_configured_instances_continues_on_failure() {
        let dir = TempDir::new().unwrap();
        let script = create_touch_socket_script(dir.path());

        let mut config = test_config_with_process("api", script.to_str().unwrap(), vec![]);
        // Add a second service with an invalid command
        config.service.insert(
            "broken".to_string(),
            ProcessConfig {
                command: "/nonexistent/binary".to_string(),
                args: vec![],
                socket: "/tmp/{name}-{id}.sock".to_string(),
                isolation: RuntimeType::Process,
                health: None,
                env: HashMap::new(),
                workdir: None,
                restart: "on-failure".to_string(),
                idle_timeout: None,
                startup_timeout: 5,
            request_timeout: 30,
                memory_limit_mb: None,
                cpu_shares: None,
                kernel: None,
                rootfs: None,
                memory_mb: 256,
                vcpus: 1,
                vsock_port: 5000,
                storage_quota_mb: None,
                storage_persist: false,
            },
        );

        // Configure both to spawn
        config.instances.insert("api".to_string(), vec!["prod".to_string()]);
        config.instances.insert("broken".to_string(), vec!["test".to_string()]);

        let hypervisor = Hypervisor::new(config);

        let (success, failed) = hypervisor.spawn_configured_instances().await;

        // One succeeded, one failed
        assert_eq!(success, 1);
        assert_eq!(failed, 1);

        // The good one should still be running
        assert!(hypervisor.is_running("api", "prod").await);
        assert!(!hypervisor.is_running("broken", "test").await);

        hypervisor.stop("api", "prod").await.ok();
    }

    // ===================
    // WEIGHTED ROUTING TESTS
    // ===================

    #[tokio::test]
    async fn test_set_weight() {
        let dir = TempDir::new().unwrap();
        let script = create_touch_socket_script(dir.path());

        let config = test_config_with_process("api", script.to_str().unwrap(), vec![]);
        let hypervisor = Hypervisor::new(config);

        hypervisor.spawn("api", "test").await.unwrap();

        // Default weight should be 100
        let info = hypervisor.get("api", "test").await.unwrap();
        assert_eq!(info.weight, 100);

        // Set weight to 50
        hypervisor.set_weight("api", "test", 50).await.unwrap();
        let info = hypervisor.get("api", "test").await.unwrap();
        assert_eq!(info.weight, 50);

        // Set weight to 0
        hypervisor.set_weight("api", "test", 0).await.unwrap();
        let info = hypervisor.get("api", "test").await.unwrap();
        assert_eq!(info.weight, 0);

        hypervisor.stop("api", "test").await.ok();
    }

    #[tokio::test]
    async fn test_set_weight_caps_at_100() {
        let dir = TempDir::new().unwrap();
        let script = create_touch_socket_script(dir.path());

        let config = test_config_with_process("api", script.to_str().unwrap(), vec![]);
        let hypervisor = Hypervisor::new(config);

        hypervisor.spawn("api", "test").await.unwrap();

        // Try to set weight > 100
        hypervisor.set_weight("api", "test", 150).await.unwrap();
        let info = hypervisor.get("api", "test").await.unwrap();
        assert_eq!(info.weight, 100); // Should be capped at 100

        hypervisor.stop("api", "test").await.ok();
    }

    #[tokio::test]
    async fn test_set_weight_nonexistent_instance() {
        let config = Config::default();
        let hypervisor = Hypervisor::new(config);

        let result = hypervisor.set_weight("api", "nonexistent", 50).await;
        assert!(result.is_err());
        assert!(result.unwrap_err().to_string().contains("not found"));
    }

    #[tokio::test]
    async fn test_list_by_process() {
        let dir = TempDir::new().unwrap();
        let script = create_touch_socket_script(dir.path());

        let config = test_config_with_process("api", script.to_str().unwrap(), vec![]);
        let hypervisor = Hypervisor::new(config);

        // Spawn multiple instances
        hypervisor.spawn("api", "prod").await.unwrap();
        hypervisor.spawn("api", "staging").await.unwrap();

        let instances = hypervisor.list_by_process("api").await;
        assert_eq!(instances.len(), 2);

        // Verify both are present
        let ids: Vec<String> = instances.iter().map(|i| i.id.id.clone()).collect();
        assert!(ids.contains(&"prod".to_string()));
        assert!(ids.contains(&"staging".to_string()));

        // Nonexistent process should return empty
        let empty = hypervisor.list_by_process("nonexistent").await;
        assert!(empty.is_empty());

        hypervisor.stop("api", "prod").await.ok();
        hypervisor.stop("api", "staging").await.ok();
    }

    #[tokio::test]
    async fn test_select_weighted_single_instance() {
        let dir = TempDir::new().unwrap();
        let script = create_touch_socket_script(dir.path());

        let config = test_config_with_process("api", script.to_str().unwrap(), vec![]);
        let hypervisor = Hypervisor::new(config);

        hypervisor.spawn("api", "prod").await.unwrap();

        // Single instance should always be selected
        for _ in 0..10 {
            let selected = hypervisor.select_weighted("api").await;
            assert!(selected.is_some());
            assert_eq!(selected.unwrap().id.id, "prod");
        }

        hypervisor.stop("api", "prod").await.ok();
    }

    #[tokio::test]
    async fn test_select_weighted_excludes_zero_weight() {
        let dir = TempDir::new().unwrap();
        let script = create_touch_socket_script(dir.path());

        let config = test_config_with_process("api", script.to_str().unwrap(), vec![]);
        let hypervisor = Hypervisor::new(config);

        hypervisor.spawn("api", "v1").await.unwrap();
        hypervisor.spawn("api", "v2").await.unwrap();

        // Set v1 weight to 0
        hypervisor.set_weight("api", "v1", 0).await.unwrap();

        // v2 should always be selected since v1 has weight 0
        for _ in 0..10 {
            let selected = hypervisor.select_weighted("api").await;
            assert!(selected.is_some());
            assert_eq!(selected.unwrap().id.id, "v2");
        }

        hypervisor.stop("api", "v1").await.ok();
        hypervisor.stop("api", "v2").await.ok();
    }

    #[tokio::test]
    async fn test_select_weighted_all_zero_weight() {
        let dir = TempDir::new().unwrap();
        let script = create_touch_socket_script(dir.path());

        let config = test_config_with_process("api", script.to_str().unwrap(), vec![]);
        let hypervisor = Hypervisor::new(config);

        hypervisor.spawn("api", "v1").await.unwrap();
        hypervisor.spawn("api", "v2").await.unwrap();

        // Set both weights to 0
        hypervisor.set_weight("api", "v1", 0).await.unwrap();
        hypervisor.set_weight("api", "v2", 0).await.unwrap();

        // No instance should be selected
        let selected = hypervisor.select_weighted("api").await;
        assert!(selected.is_none());

        hypervisor.stop("api", "v1").await.ok();
        hypervisor.stop("api", "v2").await.ok();
    }

    #[tokio::test]
    async fn test_select_weighted_no_instances() {
        let config = Config::default();
        let hypervisor = Hypervisor::new(config);

        // No instances -> None
        let selected = hypervisor.select_weighted("api").await;
        assert!(selected.is_none());
    }

    #[tokio::test]
    async fn test_select_weighted_distribution() {
        // Test that weighted selection roughly follows the weights
        let dir = TempDir::new().unwrap();
        let script = create_touch_socket_script(dir.path());

        let config = test_config_with_process("api", script.to_str().unwrap(), vec![]);
        let hypervisor = Hypervisor::new(config);

        hypervisor.spawn("api", "v1").await.unwrap();
        hypervisor.spawn("api", "v2").await.unwrap();

        // Set v1=90, v2=10
        hypervisor.set_weight("api", "v1", 90).await.unwrap();
        hypervisor.set_weight("api", "v2", 10).await.unwrap();

        let mut v1_count = 0;
        let mut v2_count = 0;
        let iterations = 1000;

        for _ in 0..iterations {
            if let Some(selected) = hypervisor.select_weighted("api").await {
                match selected.id.id.as_str() {
                    "v1" => v1_count += 1,
                    "v2" => v2_count += 1,
                    _ => panic!("Unexpected instance"),
                }
            }
        }

        // v1 should get roughly 90% of selections (with some variance)
        // Allow 15% margin for statistical variance
        let v1_ratio = v1_count as f64 / iterations as f64;
        assert!(v1_ratio > 0.75, "v1 ratio {} should be > 0.75", v1_ratio);
        assert!(v1_ratio < 0.98, "v1 ratio {} should be < 0.98", v1_ratio);

        hypervisor.stop("api", "v1").await.ok();
        hypervisor.stop("api", "v2").await.ok();
    }

    // ===================
    // DEPLOY COMMAND TESTS
    // ===================

    #[tokio::test]
    async fn test_deploy_and_wait_healthy_no_health_endpoint() {
        // When no health endpoint is configured, instance is healthy if socket exists
        let dir = TempDir::new().unwrap();
        let script = create_touch_socket_script(dir.path());

        let config = test_config_with_process("api", script.to_str().unwrap(), vec![]);
        let hypervisor = Hypervisor::new(config);

        // Deploy v2 with default weight
        let result = hypervisor.deploy_and_wait_healthy("api", "v2", 100, 5).await;
        assert!(result.is_ok(), "Deploy should succeed: {:?}", result.err());

        // Verify instance is running with correct weight
        let info = hypervisor.get("api", "v2").await.unwrap();
        assert_eq!(info.weight, 100);

        hypervisor.stop("api", "v2").await.ok();
    }

    #[tokio::test]
    async fn test_deploy_with_initial_weight() {
        // Deploy with a low initial weight (canary style)
        let dir = TempDir::new().unwrap();
        let script = create_touch_socket_script(dir.path());

        let config = test_config_with_process("api", script.to_str().unwrap(), vec![]);
        let hypervisor = Hypervisor::new(config);

        // Deploy v2 with 10% weight
        let result = hypervisor.deploy_and_wait_healthy("api", "v2", 10, 5).await;
        assert!(result.is_ok());

        // Verify weight was set
        let info = hypervisor.get("api", "v2").await.unwrap();
        assert_eq!(info.weight, 10);

        hypervisor.stop("api", "v2").await.ok();
    }

    #[tokio::test]
    async fn test_deploy_unknown_process() {
        let config = Config::default();
        let hypervisor = Hypervisor::new(config);

        let result = hypervisor.deploy_and_wait_healthy("nonexistent", "v1", 100, 5).await;
        assert!(result.is_err());
        assert!(result.unwrap_err().to_string().contains("Unknown process"));
    }

    // ===================
    // ROUTE SWAP TESTS
    // ===================

    #[tokio::test]
    async fn test_route_swap() {
        let dir = TempDir::new().unwrap();
        let script = create_touch_socket_script(dir.path());

        let config = test_config_with_process("api", script.to_str().unwrap(), vec![]);
        let hypervisor = Hypervisor::new(config);

        // Spawn v1 and v2
        hypervisor.spawn("api", "v1").await.unwrap();
        hypervisor.spawn("api", "v2").await.unwrap();

        // Initially both have weight 100
        let v1_info = hypervisor.get("api", "v1").await.unwrap();
        let v2_info = hypervisor.get("api", "v2").await.unwrap();
        assert_eq!(v1_info.weight, 100);
        assert_eq!(v2_info.weight, 100);

        // Swap traffic from v1 to v2
        let result = hypervisor.route_swap("api", "v1", "v2").await;
        assert!(result.is_ok());

        // Verify weights swapped atomically
        let v1_info = hypervisor.get("api", "v1").await.unwrap();
        let v2_info = hypervisor.get("api", "v2").await.unwrap();
        assert_eq!(v1_info.weight, 0);
        assert_eq!(v2_info.weight, 100);

        hypervisor.stop("api", "v1").await.ok();
        hypervisor.stop("api", "v2").await.ok();
    }

    #[tokio::test]
    async fn test_route_swap_from_not_found() {
        let dir = TempDir::new().unwrap();
        let script = create_touch_socket_script(dir.path());

        let config = test_config_with_process("api", script.to_str().unwrap(), vec![]);
        let hypervisor = Hypervisor::new(config);

        // Only spawn v2
        hypervisor.spawn("api", "v2").await.unwrap();

        // Try to swap from non-existent v1
        let result = hypervisor.route_swap("api", "v1", "v2").await;
        assert!(result.is_err());
        assert!(result.unwrap_err().to_string().contains("not found"));

        hypervisor.stop("api", "v2").await.ok();
    }

    #[tokio::test]
    async fn test_route_swap_to_not_found() {
        let dir = TempDir::new().unwrap();
        let script = create_touch_socket_script(dir.path());

        let config = test_config_with_process("api", script.to_str().unwrap(), vec![]);
        let hypervisor = Hypervisor::new(config);

        // Only spawn v1
        hypervisor.spawn("api", "v1").await.unwrap();

        // Try to swap to non-existent v2
        let result = hypervisor.route_swap("api", "v1", "v2").await;
        assert!(result.is_err());
        assert!(result.unwrap_err().to_string().contains("not found"));

        hypervisor.stop("api", "v1").await.ok();
    }

    #[tokio::test]
    async fn test_route_swap_weighted_routing() {
        // Verify that after route swap, only the target version receives traffic
        let dir = TempDir::new().unwrap();
        let script = create_touch_socket_script(dir.path());

        let config = test_config_with_process("api", script.to_str().unwrap(), vec![]);
        let hypervisor = Hypervisor::new(config);

        hypervisor.spawn("api", "v1").await.unwrap();
        hypervisor.spawn("api", "v2").await.unwrap();

        // Route all traffic from v1 to v2
        hypervisor.route_swap("api", "v1", "v2").await.unwrap();

        // All weighted selections should now go to v2 (v1 has weight 0)
        for _ in 0..10 {
            let selected = hypervisor.select_weighted("api").await;
            assert!(selected.is_some());
            assert_eq!(selected.unwrap().id.id, "v2");
        }

        hypervisor.stop("api", "v1").await.ok();
        hypervisor.stop("api", "v2").await.ok();
    }

    #[tokio::test]
    async fn test_blue_green_workflow() {
        // Full blue/green deployment workflow
        let dir = TempDir::new().unwrap();
        let script = create_touch_socket_script(dir.path());

        let config = test_config_with_process("api", script.to_str().unwrap(), vec![]);
        let hypervisor = Hypervisor::new(config);

        // 1. Deploy v1 (blue) initially
        hypervisor.deploy_and_wait_healthy("api", "v1", 100, 5).await.unwrap();

        // 2. Deploy v2 (green) with weight 0 (no traffic yet)
        hypervisor.deploy_and_wait_healthy("api", "v2", 0, 5).await.unwrap();

        // At this point, all traffic goes to v1
        for _ in 0..5 {
            let selected = hypervisor.select_weighted("api").await;
            assert_eq!(selected.unwrap().id.id, "v1");
        }

        // 3. Atomic swap: route all traffic from v1 to v2
        hypervisor.route_swap("api", "v1", "v2").await.unwrap();

        // Now all traffic goes to v2
        for _ in 0..5 {
            let selected = hypervisor.select_weighted("api").await;
            assert_eq!(selected.unwrap().id.id, "v2");
        }

        // 4. Stop old version v1
        hypervisor.stop("api", "v1").await.unwrap();

        // Only v2 remains
        let instances = hypervisor.list_by_process("api").await;
        assert_eq!(instances.len(), 1);
        assert_eq!(instances[0].id.id, "v2");

        hypervisor.stop("api", "v2").await.ok();
    }

    #[tokio::test]
    async fn test_canary_workflow() {
        // Full canary deployment workflow
        let dir = TempDir::new().unwrap();
        let script = create_touch_socket_script(dir.path());

        let config = test_config_with_process("api", script.to_str().unwrap(), vec![]);
        let hypervisor = Hypervisor::new(config);

        // 1. v1 is running with full traffic
        hypervisor.deploy_and_wait_healthy("api", "v1", 100, 5).await.unwrap();

        // 2. Deploy v2 with 10% traffic (canary)
        hypervisor.deploy_and_wait_healthy("api", "v2", 10, 5).await.unwrap();

        // Traffic split is now roughly 91% v1, 9% v2 (100:10 ratio)
        let v1_info = hypervisor.get("api", "v1").await.unwrap();
        let v2_info = hypervisor.get("api", "v2").await.unwrap();
        assert_eq!(v1_info.weight, 100);
        assert_eq!(v2_info.weight, 10);

        // 3. Increase v2 weight to 50%
        hypervisor.set_weight("api", "v2", 50).await.unwrap();

        // 4. Full rollout: set v1=0, v2=100
        hypervisor.set_weight("api", "v1", 0).await.unwrap();
        hypervisor.set_weight("api", "v2", 100).await.unwrap();

        // All traffic now goes to v2
        for _ in 0..5 {
            let selected = hypervisor.select_weighted("api").await;
            assert_eq!(selected.unwrap().id.id, "v2");
        }

        // 5. Stop v1
        hypervisor.stop("api", "v1").await.ok();
        hypervisor.stop("api", "v2").await.ok();
    }
}
