arcbox-core 0.4.9

//! Linux machine management.
//!
//! A "machine" is a high-level abstraction over a VM that provides
//! a Linux environment for running containers.

use crate::error::{CoreError, Result};
use crate::persistence::MachinePersistence;
use crate::vm::{SharedDirConfig, VmConfig, VmId, VmManager};
use arcbox_constants::ports::AGENT_PORT;
use arcbox_constants::virtiofs::{MOUNT_PRIVATE, MOUNT_USERS, TAG_ARCBOX, TAG_PRIVATE, TAG_USERS};
use chrono::{DateTime, Utc};
use std::collections::HashMap;
use std::net::IpAddr;
use std::path::PathBuf;
use std::sync::{Arc, RwLock};
use std::time::Duration;

/// Machine state.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum MachineState {
    /// Machine created but not started.
    Created,
    /// Machine is starting.
    Starting,
    /// Machine is running.
    Running,
    /// Machine is stopping.
    Stopping,
    /// Machine is stopped.
    Stopped,
}

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

    /// Creates a test MachineManager.
    fn test_machine_manager(data_dir: &std::path::Path) -> MachineManager {
        let vm_manager = Arc::new(VmManager::new(data_dir.join("snapshots")));
        MachineManager::new(vm_manager, data_dir.to_path_buf(), None)
    }

    #[tokio::test]
    async fn test_assign_cid_propagates_to_vm_config() {
        let temp_dir = tempdir().unwrap();
        let machine_manager = test_machine_manager(temp_dir.path());

        let name = machine_manager
            .create(MachineConfig {
                name: "cid-test".to_string(),
                ..Default::default()
            })
            .await
            .unwrap();

        let (vm_id, cid) = machine_manager.assign_cid_for_start(&name).unwrap();
        assert_eq!(cid, 3);
        assert_eq!(
            machine_manager.vm_manager.guest_cid_for_test(&vm_id),
            Some(cid)
        );
    }

    #[test]
    fn test_register_mock_machine() {
        let temp_dir = tempdir().unwrap();
        let machine_manager = test_machine_manager(temp_dir.path());

        // Register a mock machine.
        machine_manager
            .register_mock_machine("test-mock", 42)
            .unwrap();

        // Verify the machine exists.
        let machine = machine_manager
            .get("test-mock")
            .expect("machine should exist");
        assert_eq!(machine.name, "test-mock");
        assert_eq!(machine.cid, Some(42));
        assert_eq!(machine.state, MachineState::Running);
    }

    #[test]
    fn test_register_mock_machine_idempotent() {
        let temp_dir = tempdir().unwrap();
        let machine_manager = test_machine_manager(temp_dir.path());

        // Register twice should succeed (idempotent).
        machine_manager
            .register_mock_machine("test-idempotent", 10)
            .unwrap();
        machine_manager
            .register_mock_machine("test-idempotent", 20)
            .unwrap();

        // Should still have the first CID (not overwritten).
        let machine = machine_manager.get("test-idempotent").unwrap();
        assert_eq!(machine.cid, Some(10));
    }

    #[test]
    fn test_select_routable_ip_prefers_ipv4() {
        let ips = vec![
            "::1".to_string(),
            "fe80::1".to_string(),
            "2001:db8::10".to_string(),
            "10.0.2.2".to_string(),
        ];
        assert_eq!(select_routable_ip(&ips), Some("10.0.2.2".to_string()));
    }

    #[test]
    fn test_select_routable_ip_falls_back_to_global_ipv6() {
        let ips = vec![
            "::1".to_string(),
            "fe80::2".to_string(),
            "2001:db8::42".to_string(),
        ];
        assert_eq!(select_routable_ip(&ips), Some("2001:db8::42".to_string()));
    }

    /// Two concurrent `create` calls with the same name must not both succeed.
    /// One wins, the other returns `AlreadyExists`, and only one machine is
    /// registered.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn test_create_concurrent_same_name_no_duplicate() {
        let temp_dir = tempdir().unwrap();
        let machine_manager = Arc::new(test_machine_manager(temp_dir.path()));

        let name = "race-test";
        let mm1 = machine_manager.clone();
        let mm2 = machine_manager.clone();

        // `create` has no `.await` points, so without a barrier the
        // first-polled task can run to completion before the second is even
        // scheduled — defeating the contention scenario we want to cover.
        // The barrier guarantees both tasks reach the `create` call before
        // either acquires the write lock.
        let barrier = Arc::new(tokio::sync::Barrier::new(2));
        let b1 = barrier.clone();
        let b2 = barrier.clone();

        let t1 = tokio::spawn(async move {
            b1.wait().await;
            mm1.create(MachineConfig {
                name: name.to_string(),
                ..Default::default()
            })
            .await
        });
        let t2 = tokio::spawn(async move {
            b2.wait().await;
            mm2.create(MachineConfig {
                name: name.to_string(),
                ..Default::default()
            })
            .await
        });

        let r1 = t1.await.unwrap();
        let r2 = t2.await.unwrap();

        let (winner, loser) = match (r1, r2) {
            (Ok(n), Err(e)) | (Err(e), Ok(n)) => (n, e),
            (Ok(_), Ok(_)) => panic!("both creates succeeded — TOCTOU regression"),
            (Err(e1), Err(e2)) => panic!("both creates failed: {e1:?} / {e2:?}"),
        };
        assert_eq!(winner, name);
        match loser {
            CoreError::Common(ref c) if c.is_already_exists() => {}
            other => panic!("loser should be AlreadyExists, got {other:?}"),
        }

        let machines = machine_manager.list();
        assert_eq!(
            machines.len(),
            1,
            "exactly one machine should be registered"
        );
        assert_eq!(machines[0].name, name);
    }
}

/// Machine information.
#[derive(Debug, Clone)]
pub struct MachineInfo {
    /// Machine name.
    pub name: String,
    /// Machine state.
    pub state: MachineState,
    /// Underlying VM ID.
    pub vm_id: VmId,
    /// vsock CID for agent communication (assigned when VM starts).
    pub cid: Option<u32>,
    /// Number of CPUs.
    pub cpus: u32,
    /// Memory in MB.
    pub memory_mb: u64,
    /// Disk size in GB.
    pub disk_gb: u64,
    /// Kernel path.
    pub kernel: Option<String>,
    /// Kernel command line.
    pub cmdline: Option<String>,
    /// Block devices (e.g., EROFS rootfs image).
    pub block_devices: Vec<crate::vm::BlockDeviceConfig>,
    /// Distribution name (e.g., "alpine", "ubuntu").
    pub distro: Option<String>,
    /// Distribution version (e.g., "3.21", "24.04").
    pub distro_version: Option<String>,
    /// Path to the disk image.
    pub disk_path: Option<PathBuf>,
    /// Path to the SSH private key.
    pub ssh_key_path: Option<PathBuf>,
    /// Guest IP address (reported by agent via vsock).
    pub ip_address: Option<String>,
    /// Creation time.
    pub created_at: DateTime<Utc>,
}

/// Machine configuration.
#[derive(Debug, Clone)]
pub struct MachineConfig {
    /// Machine name.
    pub name: String,
    /// Number of CPUs.
    pub cpus: u32,
    /// Memory in MB.
    pub memory_mb: u64,
    /// Disk size in GB.
    pub disk_gb: u64,
    /// Kernel path.
    pub kernel: Option<String>,
    /// Kernel command line.
    pub cmdline: Option<String>,
    /// Block devices (e.g., EROFS rootfs image).
    pub block_devices: Vec<crate::vm::BlockDeviceConfig>,
    /// Distribution name (e.g., "alpine", "ubuntu").
    pub distro: Option<String>,
    /// Distribution version (e.g., "3.21", "24.04").
    pub distro_version: Option<String>,
    /// macOS hypervisor backend for this machine.
    ///
    /// `Hv` runs ArcBox's custom HV-framework VMM (fast path for
    /// `linux/arm64` workloads); `Vz` runs Apple's
    /// Virtualization.framework managed execution (required for Rosetta).
    /// Defaults to `Hv` so the existing single-VM behavior is preserved.
    pub backend: arcbox_vmm::VmBackend,
    /// Whether to expose Apple Rosetta inside the guest for `linux/amd64`
    /// translation.
    ///
    /// Only honored when [`Self::backend`] is `Vz`; the HV path silently
    /// drops it because Hypervisor.framework does not host the Rosetta
    /// share. Defaults to `false`.
    pub enable_rosetta: bool,
}

impl Default for MachineConfig {
    fn default() -> Self {
        Self {
            name: "default".to_string(),
            cpus: arcbox_hypervisor::default_vm_cpu_count(),
            memory_mb: 4096,
            disk_gb: 50,
            kernel: None,
            cmdline: None,
            block_devices: Vec::new(),
            distro: None,
            distro_version: None,
            backend: arcbox_vmm::VmBackend::Hv,
            enable_rosetta: false,
        }
    }
}

/// Machine manager.
pub struct MachineManager {
    machines: RwLock<HashMap<String, MachineInfo>>,
    vm_manager: Arc<VmManager>,
    persistence: MachinePersistence,
    /// Data directory for `VirtioFS` sharing.
    data_dir: PathBuf,
    /// Machine-specific directory (`data_dir/machines`/).
    machines_dir: PathBuf,
    /// Shared DNS hosts table from NetworkManager, passed to VMM on start.
    shared_dns_hosts: Option<std::sync::Arc<arcbox_dns::LocalHostsTable>>,
}

impl MachineManager {
    /// Creates a new machine manager.
    #[must_use]
    pub fn new(
        vm_manager: Arc<VmManager>,
        data_dir: PathBuf,
        shared_dns_hosts: Option<std::sync::Arc<arcbox_dns::LocalHostsTable>>,
    ) -> Self {
        let machines_dir = data_dir.join("machines");
        let persistence = MachinePersistence::new(&machines_dir);

        // Create the default shared directory config for VirtioFS.
        // "arcbox" shares the data_dir; "users" shares /Users for transparent paths.
        let mut shared_dirs = vec![SharedDirConfig::new(
            data_dir.to_string_lossy().to_string(),
            TAG_ARCBOX,
        )];
        let users_dir = std::path::Path::new(MOUNT_USERS);
        if users_dir.is_dir() {
            shared_dirs.push(SharedDirConfig::new(MOUNT_USERS, TAG_USERS));
        }
        let private_dir = std::path::Path::new(MOUNT_PRIVATE);
        if private_dir.is_dir() {
            shared_dirs.push(SharedDirConfig::new(MOUNT_PRIVATE, TAG_PRIVATE));
        }

        // Load persisted machines
        let mut machines = HashMap::new();
        for persisted in persistence.load_all() {
            let needs_recovery = persisted.state.needs_recovery();
            if needs_recovery {
                tracing::warn!(
                    "Machine '{}' was running when daemon stopped — marking as stopped",
                    persisted.name
                );
            }

            // Reconstruct VmConfig from persisted data.
            let vm_config = VmConfig {
                cpus: persisted.cpus,
                memory_mb: persisted.memory_mb,
                kernel: persisted.kernel.clone(),
                cmdline: persisted.cmdline.clone(),
                shared_dirs: shared_dirs.clone(),
                block_devices: persisted.block_devices.clone(),
                ..Default::default()
            };

            // Try to create the underlying VM
            if let Ok(vm_id) = vm_manager.create(vm_config) {
                let info = MachineInfo {
                    name: persisted.name.clone(),
                    state: persisted.state.into(),
                    vm_id,
                    cid: None, // Will be assigned when VM starts
                    cpus: persisted.cpus,
                    memory_mb: persisted.memory_mb,
                    disk_gb: persisted.disk_gb,
                    kernel: persisted.kernel.clone(),
                    cmdline: persisted.cmdline,
                    block_devices: persisted.block_devices.clone(),
                    distro: persisted.distro.clone(),
                    distro_version: persisted.distro_version.clone(),
                    disk_path: persisted.disk_path.clone().map(PathBuf::from),
                    ssh_key_path: persisted.ssh_key_path.clone().map(PathBuf::from),
                    ip_address: persisted.ip_address.clone(),
                    created_at: persisted.created_at,
                };
                machines.insert(persisted.name.clone(), info);
            }

            // Persist the corrected state regardless of whether VM recreation
            // succeeded — a stale Running on disk must not survive reload.
            if needs_recovery {
                if let Err(e) = persistence.update_state(&persisted.name, MachineState::Stopped) {
                    tracing::warn!(
                        "Failed to persist corrected state for '{}': {}",
                        persisted.name,
                        e
                    );
                }
            }
        }

        tracing::info!("Loaded {} persisted machines", machines.len());

        Self {
            machines: RwLock::new(machines),
            vm_manager,
            persistence,
            data_dir,
            machines_dir,
            shared_dns_hosts,
        }
    }

    /// Creates a new machine.
    ///
    /// Sets up EROFS rootfs (read-only, /dev/vda) and a Btrfs data disk
    /// (/dev/vdb) with block device and `VirtioFS` sharing configured.
    ///
    /// When `config.distro` is set, also resolves and downloads a distro
    /// rootfs tarball and generates an SSH key pair.
    ///
    /// # Errors
    ///
    /// Returns an error if the machine cannot be created.
    pub async fn create(&self, config: MachineConfig) -> Result<String> {
        // Hold the write lock for the entire create operation to prevent TOCTOU
        // races: without this, two concurrent creates with the same name could
        // both pass the existence check before either inserts. `create` is rare
        // and user-driven, so the alternative (insert a `Creating` sentinel,
        // drop the lock for I/O, then finalize/rollback) is not worth its
        // orphan-state failure mode.
        let mut machines = self.machines.write().map_err(|_| CoreError::LockPoisoned)?;

        if machines.contains_key(&config.name) {
            return Err(CoreError::already_exists(config.name));
        }

        let machine_dir = self.machines_dir.join(&config.name);
        std::fs::create_dir_all(&machine_dir)?;

        // Set up shared directories for VirtioFS.
        // "arcbox" tag provides internal data (boot assets, logs, runtime).
        // "users" tag shares /Users so macOS paths work transparently in guest
        // (e.g. `docker run -v /Users/foo/project:/app` just works).
        let mut shared_dirs = vec![SharedDirConfig::new(
            self.data_dir.to_string_lossy().to_string(),
            TAG_ARCBOX,
        )];
        let users_dir = std::path::Path::new(MOUNT_USERS);
        if users_dir.is_dir() {
            shared_dirs.push(SharedDirConfig::new(MOUNT_USERS, TAG_USERS));
        }
        let private_dir = std::path::Path::new(MOUNT_PRIVATE);
        if private_dir.is_dir() {
            shared_dirs.push(SharedDirConfig::new(MOUNT_PRIVATE, TAG_PRIVATE));
        }

        // Create underlying VM
        let vm_config = VmConfig {
            cpus: config.cpus,
            memory_mb: config.memory_mb,
            kernel: config.kernel.clone(),
            cmdline: config.cmdline.clone(),
            shared_dirs,
            block_devices: config.block_devices.clone(),
            rosetta: config.enable_rosetta,
            backend: config.backend,
            ..Default::default()
        };
        let vm_id = self.vm_manager.create(vm_config)?;

        let info = MachineInfo {
            name: config.name.clone(),
            state: MachineState::Created,
            vm_id,
            cid: None,
            cpus: config.cpus,
            memory_mb: config.memory_mb,
            disk_gb: config.disk_gb,
            kernel: config.kernel,
            cmdline: config.cmdline,
            block_devices: config.block_devices,
            distro: config.distro,
            distro_version: config.distro_version,
            disk_path: None,
            ssh_key_path: None,
            ip_address: None,
            created_at: Utc::now(),
        };

        // Persist the machine config
        self.persistence.save(&info)?;

        machines.insert(config.name.clone(), info);

        Ok(config.name)
    }

    /// Starts a machine.
    ///
    /// For machine VMs with a distro, this also waits for the guest agent to
    /// become ready and discovers the guest IP address via vsock.
    ///
    /// # Errors
    ///
    /// Returns an error if the machine cannot be started.
    pub async fn start(&self, name: &str) -> Result<()> {
        let (vm_id, cid) = self.assign_cid_for_start(name)?;

        // Check if this is a distro-based machine VM.
        let is_machine_vm = self
            .machines
            .read()
            .map_err(|_| CoreError::LockPoisoned)?
            .get(name)
            .and_then(|m| m.distro.as_ref())
            .is_some();

        // Start underlying VM
        self.vm_manager
            .start(&vm_id, self.shared_dns_hosts.clone())?;

        // Update machine state
        {
            let mut machines = self.machines.write().map_err(|_| CoreError::LockPoisoned)?;

            if let Some(machine) = machines.get_mut(name) {
                machine.state = MachineState::Running;
                machine.cid = Some(cid);
                machine.ip_address = None;

                tracing::info!("Machine '{}' started with CID {}", name, cid);
            }
        }

        // Update persisted state (single read-modify-write)
        if let Err(e) = self.persistence.update(name, |m| {
            m.state = MachineState::Running.into();
            m.ip_address = None;
        }) {
            tracing::warn!("Failed to persist state for machine '{}': {}", name, e);
        }

        // For machine VMs, wait for agent readiness and discover IP.
        if is_machine_vm {
            self.wait_for_machine_ready(name).await.map_err(|e| {
                CoreError::Machine(format!(
                    "Machine '{name}' started but readiness check failed: {e}"
                ))
            })?;
        }

        Ok(())
    }

    /// Waits for the guest agent to become ready and discovers the IP address.
    ///
    /// Polls the agent via vsock with exponential backoff. Once the agent
    /// responds, queries `SystemInfo` to get the guest IP.
    ///
    /// Runs the probe loop on a blocking thread to avoid tokio reactor
    /// stalls from rapid socketpair fd teardown (same rationale as
    /// `wait_for_agent` in `vm_lifecycle`).
    async fn wait_for_machine_ready(&self, name: &str) -> Result<()> {
        const MAX_ATTEMPTS: u32 = 20;
        const INITIAL_DELAY_MS: u64 = 500;
        const MAX_DELAY_MS: u64 = 3000;

        tracing::info!("Waiting for machine '{}' agent to become ready...", name);

        // block_in_place is used here (instead of spawn_blocking) because
        // MachineManager is not Clone/Arc at this call site. block_in_place
        // is acceptable: the total blocking time is bounded by
        // MAX_ATTEMPTS * MAX_DELAY_MS ≈ 60s, and the blocking RPC uses
        // BlockingVsockTransport (libc::poll) — no tokio reactor interaction.
        let probe_result: Result<String> = tokio::task::block_in_place(|| {
            let mut delay_ms = INITIAL_DELAY_MS;

            for attempt in 1..=MAX_ATTEMPTS {
                std::thread::sleep(std::time::Duration::from_millis(delay_ms));

                match self.connect_agent(name) {
                    Ok(mut agent) if agent.is_blocking() => match agent.ping_blocking() {
                        Ok(resp) => {
                            tracing::debug!(
                                "Machine '{}' agent reachable (version: {}, attempt {})",
                                name,
                                resp.version,
                                attempt,
                            );
                            match agent.get_system_info_blocking() {
                                Ok(info) => {
                                    if let Some(ip) = select_routable_ip(&info.ip_addresses) {
                                        return Ok(ip);
                                    }
                                    tracing::trace!(
                                        "Machine '{}' no routable IP yet (attempt {})",
                                        name,
                                        attempt,
                                    );
                                }
                                Err(e) => tracing::trace!(
                                    "Machine '{}' get_system_info failed (attempt {attempt}): {e}",
                                    name,
                                ),
                            }
                        }
                        Err(e) => tracing::trace!(
                            "Machine '{}' ping failed (attempt {attempt}): {e}",
                            name,
                        ),
                    },
                    Ok(_agent) => {
                        // Async transport (VZ/Linux) — skip in blocking context.
                        tracing::trace!(
                            "Machine '{}' async transport in blocking probe (attempt {})",
                            name,
                            attempt,
                        );
                    }
                    Err(e) => tracing::trace!(
                        "Machine '{}' connect failed (attempt {attempt}): {e}",
                        name,
                    ),
                }

                delay_ms = (delay_ms * 3 / 2).min(MAX_DELAY_MS);
            }

            Err(CoreError::Machine(format!(
                "Machine '{name}' agent did not report a routable IP within timeout"
            )))
        });

        let ip = probe_result?;

        // Back on async context — update state.
        {
            let mut machines = self.machines.write().map_err(|_| CoreError::LockPoisoned)?;
            if let Some(machine) = machines.get_mut(name) {
                machine.ip_address = Some(ip.clone());
            }
        }
        if let Err(e) = self.persistence.update_ip(name, Some(&ip)) {
            tracing::warn!("Failed to persist IP for machine '{}': {}", name, e);
        }
        tracing::info!("Machine '{}' ready with IP {}", name, ip);
        Ok(())
    }

    fn assign_cid_for_start(&self, name: &str) -> Result<(VmId, u32)> {
        let (vm_id, running_count) = {
            let machines = self.machines.read().map_err(|_| CoreError::LockPoisoned)?;

            let machine = machines
                .get(name)
                .ok_or_else(|| CoreError::not_found(name.to_string()))?;

            if machine.state == MachineState::Running {
                return Err(CoreError::invalid_state(format!(
                    "machine '{name}' is already running"
                )));
            }

            if machine.state == MachineState::Starting || machine.state == MachineState::Stopping {
                return Err(CoreError::invalid_state(format!(
                    "machine '{name}' is in transition state"
                )));
            }

            // Count running machines. CIDs 0, 1 are reserved, 2 is the host. We start from 3.
            let running_count = machines
                .values()
                .filter(|m| m.state == MachineState::Running && m.cid.is_some())
                .count() as u32;

            (machine.vm_id.clone(), running_count)
        };

        let cid = 3 + running_count;
        self.vm_manager.set_guest_cid(&vm_id, cid)?;

        Ok((vm_id, cid))
    }

    /// Returns a reference to the underlying VM manager.
    #[must_use]
    pub fn vm_manager(&self) -> &VmManager {
        &self.vm_manager
    }

    /// Returns the vmnet bridge interface name for a machine's VM.
    ///
    /// Only available when the `vmnet` feature is enabled and the VM is running.
    #[cfg(all(target_os = "macos", feature = "vmnet"))]
    pub fn vmnet_bridge_name(&self, name: &str) -> Option<String> {
        let machines = self.machines.read().ok()?;
        let machine = machines.get(name)?;
        self.vm_manager.vmnet_bridge_name(&machine.vm_id)
    }

    /// Returns the bridge NIC MAC address for a machine's VM.
    pub fn bridge_mac(&self, name: &str) -> Option<String> {
        let machines = self.machines.read().ok()?;
        let machine = machines.get(name)?;
        Some(crate::vm::bridge_nic_mac_for_vm_id(&machine.vm_id))
    }

    /// Gets the vsock CID for a running machine.
    #[must_use]
    pub fn get_cid(&self, name: &str) -> Option<u32> {
        self.machines.read().ok()?.get(name)?.cid
    }

    /// Connects to the agent on a running machine.
    ///
    /// Returns an `AgentClient` that can be used to communicate with the
    /// guest agent for container operations.
    ///
    /// # Errors
    /// Returns an error if the machine is not found, not running, or connection fails.
    #[cfg(target_os = "macos")]
    pub fn connect_agent(&self, name: &str) -> Result<crate::agent_client::AgentClient> {
        use crate::agent_client::AgentClient;
        let cid = self
            .get_cid(name)
            .ok_or_else(|| CoreError::invalid_state("CID not assigned"))?;
        let fd = self.connect_vsock_port(name, AGENT_PORT)?;
        AgentClient::from_fd(cid, fd)
    }

    /// Connects to a vsock port on a running machine (macOS).
    ///
    /// This is a generic helper used by agent and guest runtime proxy paths.
    #[cfg(target_os = "macos")]
    pub fn connect_vsock_port(&self, name: &str, port: u32) -> Result<std::os::unix::io::RawFd> {
        let machines = self.machines.read().map_err(|_| CoreError::LockPoisoned)?;

        let machine = machines
            .get(name)
            .ok_or_else(|| CoreError::not_found(name.to_string()))?;

        if machine.state != MachineState::Running {
            return Err(CoreError::invalid_state(format!(
                "machine '{name}' is not running"
            )));
        }

        self.vm_manager.connect_vsock(&machine.vm_id, port)
    }

    /// Connects to the agent on a running machine (Linux).
    #[cfg(target_os = "linux")]
    pub fn connect_agent(&self, name: &str) -> Result<crate::agent_client::AgentClient> {
        use crate::agent_client::AgentClient;

        let machines = self.machines.read().map_err(|_| CoreError::LockPoisoned)?;

        let machine = machines
            .get(name)
            .ok_or_else(|| CoreError::not_found(name.to_string()))?;

        if machine.state != MachineState::Running {
            return Err(CoreError::invalid_state(format!(
                "machine '{}' is not running",
                name
            )));
        }

        let cid = machine
            .cid
            .ok_or_else(|| CoreError::invalid_state("CID not assigned"))?;

        // On Linux, AgentClient connects directly via AF_VSOCK
        Ok(AgentClient::new(cid))
    }

    /// Connects to a vsock port on a running machine (Linux).
    #[cfg(target_os = "linux")]
    pub fn connect_vsock_port(&self, name: &str, port: u32) -> Result<std::os::unix::io::RawFd> {
        let machines = self.machines.read().map_err(|_| CoreError::LockPoisoned)?;

        let machine = machines
            .get(name)
            .ok_or_else(|| CoreError::not_found(name.to_string()))?;

        if machine.state != MachineState::Running {
            return Err(CoreError::invalid_state(format!(
                "machine '{}' is not running",
                name
            )));
        }

        self.vm_manager.connect_vsock(&machine.vm_id, port)
    }

    /// Reads serial console output for a running machine (macOS only).
    #[cfg(target_os = "macos")]
    pub fn read_console_output(&self, name: &str) -> Result<String> {
        let machines = self.machines.read().map_err(|_| CoreError::LockPoisoned)?;

        let machine = machines
            .get(name)
            .ok_or_else(|| CoreError::not_found(name.to_string()))?;

        if machine.state != MachineState::Running {
            return Err(CoreError::invalid_state(format!(
                "machine '{name}' is not running"
            )));
        }

        self.vm_manager.read_console_output(&machine.vm_id)
    }

    /// Reads agent log output (hvc1) for a running machine (macOS only).
    #[cfg(target_os = "macos")]
    pub fn read_agent_log_output(&self, name: &str) -> Result<String> {
        let machines = self.machines.read().map_err(|_| CoreError::LockPoisoned)?;

        let machine = machines
            .get(name)
            .ok_or_else(|| CoreError::not_found(name.to_string()))?;

        if machine.state != MachineState::Running {
            return Err(CoreError::invalid_state(format!(
                "machine '{name}' is not running"
            )));
        }

        self.vm_manager.read_agent_log_output(&machine.vm_id)
    }

    /// Stops a machine.
    ///
    /// # Errors
    ///
    /// Returns an error if the machine cannot be stopped.
    pub fn stop(&self, name: &str) -> Result<()> {
        let mut machines = self.machines.write().map_err(|_| CoreError::LockPoisoned)?;

        let machine = machines
            .get_mut(name)
            .ok_or_else(|| CoreError::not_found(name.to_string()))?;

        if machine.state != MachineState::Running {
            return Err(CoreError::invalid_state(format!(
                "machine '{name}' is not running"
            )));
        }

        // Stop underlying VM
        #[cfg(target_os = "macos")]
        self.vm_manager
            .force_stop_without_hypervisor(&machine.vm_id)?;
        #[cfg(not(target_os = "macos"))]
        self.vm_manager.stop(&machine.vm_id)?;

        machine.state = MachineState::Stopped;
        machine.cid = None;

        // Update persisted state
        if let Err(e) = self.persistence.update_state(name, MachineState::Stopped) {
            tracing::warn!(
                "Failed to persist stopped state for machine '{}': {}",
                name,
                e
            );
        }

        Ok(())
    }

    /// Attempts graceful machine shutdown via guest ACPI stop request.
    ///
    /// Returns `Ok(true)` if the machine stopped, `Ok(false)` if graceful
    /// shutdown timed out or is unavailable.
    pub fn graceful_stop(&self, name: &str, timeout: Duration) -> Result<bool> {
        let vm_id = {
            let mut machines = self.machines.write().map_err(|_| CoreError::LockPoisoned)?;

            let machine = machines
                .get_mut(name)
                .ok_or_else(|| CoreError::not_found(name.to_string()))?;

            if machine.state != MachineState::Running {
                return Err(CoreError::invalid_state(format!(
                    "machine '{name}' is not running"
                )));
            }

            machine.state = MachineState::Stopping;
            machine.vm_id.clone()
        };

        match self.vm_manager.graceful_stop(&vm_id, timeout) {
            Ok(true) => {
                let mut machines = self.machines.write().map_err(|_| CoreError::LockPoisoned)?;

                let machine = machines
                    .get_mut(name)
                    .ok_or_else(|| CoreError::not_found(name.to_string()))?;
                machine.state = MachineState::Stopped;
                machine.cid = None;

                if let Err(e) = self.persistence.update_state(name, MachineState::Stopped) {
                    tracing::warn!(
                        "Failed to persist stopped state for machine '{}': {}",
                        name,
                        e
                    );
                }
                Ok(true)
            }
            Ok(false) => {
                if let Ok(mut machines) = self.machines.write() {
                    if let Some(machine) = machines.get_mut(name) {
                        machine.state = MachineState::Running;
                    }
                }
                Ok(false)
            }
            Err(e) => {
                if let Ok(mut machines) = self.machines.write() {
                    if let Some(machine) = machines.get_mut(name) {
                        machine.state = MachineState::Running;
                    }
                }
                Err(e)
            }
        }
    }

    /// Gets machine information.
    #[must_use]
    pub fn get(&self, name: &str) -> Option<MachineInfo> {
        self.machines.read().ok()?.get(name).cloned()
    }

    /// Lists all machines.
    #[must_use]
    pub fn list(&self) -> Vec<MachineInfo> {
        self.machines
            .read()
            .map(|m| m.values().cloned().collect())
            .unwrap_or_default()
    }

    /// Removes a machine and all associated artifacts (disk, SSH keys, config).
    ///
    /// # Errors
    ///
    /// Returns an error if the machine cannot be removed.
    pub fn remove(&self, name: &str, force: bool) -> Result<()> {
        let mut machines = self.machines.write().map_err(|_| CoreError::LockPoisoned)?;

        let machine = machines
            .get(name)
            .ok_or_else(|| CoreError::not_found(name.to_string()))?;

        // Check if machine is running
        if machine.state == MachineState::Running && !force {
            return Err(CoreError::invalid_state(
                "cannot remove running machine (use --force)".to_string(),
            ));
        }

        // Stop if running and force is set
        if machine.state == MachineState::Running {
            let vm_id = machine.vm_id.clone();
            drop(machines); // Release lock before stopping
            self.vm_manager.stop(&vm_id)?;
            machines = self.machines.write().map_err(|_| CoreError::LockPoisoned)?;
        }

        // Get VM ID before removing from map.
        let vm_id = {
            let m = machines
                .get(name)
                .ok_or_else(|| CoreError::not_found(name.to_string()))?;
            m.vm_id.clone()
        };

        // Remove from VM manager
        self.vm_manager.remove(&vm_id)?;

        // Remove from machines map
        machines.remove(name);

        // Remove persisted config (removes entire machine directory including SSH keys).
        // This must succeed — if it doesn't, the machine will reappear on daemon
        // restart even though VM and in-memory state are already gone.
        self.persistence.remove(name)?;

        tracing::info!("Removed machine '{}'", name);
        Ok(())
    }

    /// Takes the inbound listener manager from a running machine's VM (Darwin only).
    ///
    /// Returns `None` if the machine is not found, not running, or the manager
    /// has already been taken.
    #[cfg(target_os = "macos")]
    pub fn take_inbound_listener_manager(
        &self,
        name: &str,
    ) -> Option<arcbox_net::darwin::inbound_relay::InboundListenerManager> {
        let vm_id = {
            let machines = self.machines.read().ok()?;
            let machine = machines.get(name)?;
            if machine.state != MachineState::Running {
                return None;
            }
            machine.vm_id.clone()
        };
        self.vm_manager.take_inbound_listener_manager(&vm_id)
    }

    /// Registers a mock machine for testing purposes.
    ///
    /// This method creates a machine entry without creating an actual VM.
    /// The machine will be in Running state with a mock CID.
    ///
    /// # Note
    /// This is intended for unit testing only and should not be used in production.
    pub fn register_mock_machine(&self, name: &str, cid: u32) -> Result<()> {
        let mut machines = self.machines.write().map_err(|_| CoreError::LockPoisoned)?;

        if machines.contains_key(name) {
            return Ok(()); // Already registered
        }

        let info = MachineInfo {
            name: name.to_string(),
            state: MachineState::Running,
            vm_id: VmId::new(), // Fake VM ID
            cid: Some(cid),
            cpus: arcbox_hypervisor::default_vm_cpu_count(),
            memory_mb: 4096,
            disk_gb: 50,
            kernel: None,
            cmdline: None,
            block_devices: Vec::new(),
            distro: None,
            distro_version: None,
            disk_path: None,
            ssh_key_path: None,
            ip_address: None,
            created_at: Utc::now(),
        };

        machines.insert(name.to_string(), info);
        tracing::debug!("Registered mock machine '{}' with CID {}", name, cid);
        Ok(())
    }
}

fn select_routable_ip(ips: &[String]) -> Option<String> {
    let mut ipv6_candidate = None;

    for ip in ips {
        let Ok(addr) = ip.parse::<IpAddr>() else {
            continue;
        };
        if addr.is_loopback() || addr.is_multicast() || addr.is_unspecified() {
            continue;
        }

        match addr {
            IpAddr::V4(v4) => return Some(v4.to_string()),
            IpAddr::V6(v6) => {
                if v6.is_unicast_link_local() {
                    continue;
                }
                if ipv6_candidate.is_none() {
                    ipv6_candidate = Some(v6.to_string());
                }
            }
        }
    }

    ipv6_candidate
}