a3s_box_runtime/local_execution/

vm_backend.rs

//! Production local execution backend backed by [`crate::VmManager`].

#[path = "vm_sandbox.rs"]
mod sandbox;

use std::path::{Path, PathBuf};
use std::sync::Arc;
#[cfg(unix)]
use std::time::Duration;

use a3s_box_core::{
    EventEmitter, ExecutionBackend, ExecutionId, ExecutionManagerError, ExecutionManagerResult,
    ExecutionState, KillOutcome, DEFAULT_SHUTDOWN_TIMEOUT_MS,
};
use async_trait::async_trait;
use dashmap::mapref::entry::Entry;
use dashmap::DashMap;
use tokio::sync::Mutex;

use super::resources::ExecutionResourceGuard;
use super::vm_process::{locate_microvm_process, LocatedProcess};
#[cfg(target_os = "linux")]
use super::TransientRegistryAuthBroker;
use super::{
    LocalExecutionBackend, LocalExecutionHandle, LocalExecutionObservation,
    LocalExecutionTermination,
};
use crate::vm::{TERMINAL_EXIT_POLL_INTERVAL, TERMINAL_EXIT_POLL_TIMEOUT};
use crate::{
    BoxRecord, ManagedExecutionMetadata, ManagedExecutionOperation, ManagedExecutionState,
    VmManager,
};

type SharedVm = Arc<Mutex<VmManager>>;

/// Runtime adapter that owns live [`VmManager`] handles and reconstructs them
/// from durable runtime evidence after a control-plane restart.
#[derive(Clone)]
pub struct VmLocalExecutionBackend {
    home_dir: PathBuf,
    managers: Arc<DashMap<String, SharedVm>>,
    pull_progress_fn: Option<crate::PullProgressFn>,
    #[cfg(target_os = "linux")]
    transient_registry_auth: Option<TransientRegistryAuthBroker>,
}

impl VmLocalExecutionBackend {
    pub fn new(home_dir: impl Into<PathBuf>) -> Self {
        Self {
            home_dir: home_dir.into(),
            managers: Arc::new(DashMap::new()),
            pull_progress_fn: None,
            #[cfg(target_os = "linux")]
            transient_registry_auth: None,
        }
    }

    pub fn with_pull_progress_fn(mut self, pull_progress_fn: crate::PullProgressFn) -> Self {
        self.pull_progress_fn = Some(pull_progress_fn);
        self
    }

    #[cfg(target_os = "linux")]
    pub(crate) fn with_transient_registry_auth(
        mut self,
        broker: TransientRegistryAuthBroker,
    ) -> Self {
        self.transient_registry_auth = Some(broker);
        self
    }

    pub fn home_dir(&self) -> &Path {
        &self.home_dir
    }

    fn metadata<'a>(
        &self,
        record: &'a BoxRecord,
    ) -> ExecutionManagerResult<&'a ManagedExecutionMetadata> {
        self.metadata_for_route(record, crate::ManagedRuntimeRoute::BoxVm)
    }

    fn metadata_for_route<'a>(
        &self,
        record: &'a BoxRecord,
        expected_route: crate::ManagedRuntimeRoute,
    ) -> ExecutionManagerResult<&'a ManagedExecutionMetadata> {
        uuid::Uuid::parse_str(&record.id).map_err(|error| {
            ExecutionManagerError::Internal(format!(
                "managed execution has an invalid internal ID {}: {error}",
                record.id
            ))
        })?;
        let expected_box_dir = self.home_dir.join("boxes").join(&record.id);
        if record.box_dir != expected_box_dir {
            return Err(ExecutionManagerError::Internal(format!(
                "managed execution {} has an unexpected host directory {}",
                record.id,
                record.box_dir.display()
            )));
        }
        let metadata = record.managed_execution.as_ref().ok_or_else(|| {
            ExecutionManagerError::Internal(format!(
                "execution {} lost managed lifecycle metadata",
                record.id
            ))
        })?;
        metadata
            .validate()
            .map_err(|error| ExecutionManagerError::Internal(error.to_string()))?;
        let resolved_route = super::router::resolved_runtime_route(record)?;
        if resolved_route != expected_route {
            return Err(ExecutionManagerError::Internal(format!(
                "managed execution {} is pinned to {:?}, not {:?}",
                record.id, resolved_route, expected_route
            )));
        }
        if record.isolation != metadata.request.config.isolation {
            return Err(ExecutionManagerError::Internal(format!(
                "managed execution {} has inconsistent isolation metadata",
                record.id
            )));
        }
        Ok(metadata)
    }

    fn new_manager(&self, record: &BoxRecord) -> ExecutionManagerResult<VmManager> {
        let metadata = self.metadata(record)?;
        self.new_manager_from_metadata(record, metadata)
    }

    pub(super) fn new_oci_preparation_manager(
        &self,
        record: &BoxRecord,
    ) -> ExecutionManagerResult<VmManager> {
        let metadata = self.metadata_for_route(record, crate::ManagedRuntimeRoute::OciSdk)?;
        self.new_manager_from_metadata(record, metadata)
    }

    fn new_manager_from_metadata(
        &self,
        record: &BoxRecord,
        metadata: &ManagedExecutionMetadata,
    ) -> ExecutionManagerResult<VmManager> {
        let mut config = metadata.request.config.clone();
        // Network connect/disconnect is intentionally mutable while an
        // execution is inactive. The managed creation request remains the
        // immutable idempotency identity, so apply the current record-level
        // network selection when constructing each runtime generation.
        config.network = record.network_mode.clone();
        if let Some(shm_size) = metadata.request.policy.shm_size {
            let has_shared_memory_mount = config
                .tmpfs
                .iter()
                .any(|entry| entry.split(':').next() == Some("/dev/shm"));
            if !has_shared_memory_mount {
                config.tmpfs.push(format!("/dev/shm:size={shm_size}"));
            }
        }
        let mut manager = VmManager::with_box_id(config, EventEmitter::new(256), record.id.clone());
        manager.home_dir = self.home_dir.clone();
        manager.set_healthcheck_disabled(metadata.request.policy.healthcheck_disabled);
        if let Some(pull_progress_fn) = self.pull_progress_fn.clone() {
            manager.set_pull_progress_fn(pull_progress_fn);
        }
        manager.anonymous_volumes = record.anonymous_volumes.clone();
        manager.set_log_config(record.log_config.clone());
        manager.resolved_execution_plan = Some(metadata.plan.clone());
        manager.managed_secret_root = metadata.request.policy.managed_secret_root.clone();
        Ok(manager)
    }

    #[cfg(target_os = "linux")]
    fn claim_transient_registry_auth_for_boot(&self, manager: &mut VmManager) {
        manager.transient_registry_auth = self
            .transient_registry_auth
            .as_ref()
            .and_then(|broker| broker.take(manager.box_id()));
    }

    fn manager(&self, execution_id: &str) -> Option<SharedVm> {
        self.managers
            .get(execution_id)
            .map(|entry| Arc::clone(entry.value()))
    }

    fn remove_manager(&self, execution_id: &str, expected: &SharedVm) {
        if let Entry::Occupied(entry) = self.managers.entry(execution_id.to_string()) {
            if Arc::ptr_eq(entry.get(), expected) {
                entry.remove();
            }
        }
    }

    async fn handle_from_manager(
        &self,
        record: &BoxRecord,
        manager: &VmManager,
    ) -> ExecutionManagerResult<LocalExecutionHandle> {
        let execution_id = execution_id(record)?;
        let pid = manager.pid().await.ok_or_else(|| {
            ExecutionManagerError::Internal(format!(
                "runtime returned no host PID for {execution_id}"
            ))
        })?;
        let pid_start_time = crate::process::pid_start_time(pid);
        #[cfg(target_os = "linux")]
        if pid_start_time.is_none() {
            return Err(ExecutionManagerError::NotFound(execution_id));
        }
        if !crate::process::is_process_alive_with_identity(pid, pid_start_time) {
            return Err(ExecutionManagerError::NotFound(execution_id));
        }
        let exec_socket_path = manager
            .exec_socket_path()
            .map(Path::to_path_buf)
            .ok_or_else(|| {
                ExecutionManagerError::Internal(format!(
                    "runtime returned no exec socket for {}",
                    record.id
                ))
            })?;
        let anonymous_volumes = if manager.anonymous_volumes().is_empty() {
            self.anonymous_volumes_for_record(record).await
        } else {
            manager.anonymous_volumes().to_vec()
        };
        Ok(LocalExecutionHandle {
            started_at: record.started_at.unwrap_or_else(chrono::Utc::now),
            pid: Some(pid),
            pid_start_time,
            exec_socket_path,
            console_log: record.box_dir.join("logs/console.log"),
            anonymous_volumes,
            oci_runtime: None,
        })
    }

    async fn inspect_registered(
        &self,
        record: &BoxRecord,
        shared: SharedVm,
    ) -> ExecutionManagerResult<LocalExecutionObservation> {
        let mut manager = shared.lock().await;
        let preserve_rootfs = should_force_rootfs_preservation(record)?;
        let exit_code = manager
            .try_wait_exit()
            .await
            .map_err(|error| runtime_error("inspect", record, error))?;
        let mut state = manager.state().await;
        let terminal = exit_code.is_some() || state == crate::BoxState::Stopped;
        if terminal {
            return self
                .finish_registered_terminal(record, &shared, manager, preserve_rootfs, exit_code)
                .await;
        }

        if state == crate::BoxState::Created {
            if manager.has_exited().await {
                return self
                    .finish_registered_terminal(record, &shared, manager, preserve_rootfs, None)
                    .await;
            }
            if !self.promote_if_ready(record, &mut manager).await {
                return Ok(LocalExecutionObservation {
                    state: ExecutionState::Creating,
                    handle: None,
                    exit_code: None,
                });
            }
            state = manager.state().await;
        }

        if !manager
            .health_check()
            .await
            .map_err(|error| runtime_error("inspect", record, error))?
        {
            return self
                .finish_registered_terminal(record, &shared, manager, preserve_rootfs, None)
                .await;
        }

        if state != crate::BoxState::Ready
            && state != crate::BoxState::Busy
            && state != crate::BoxState::Compacting
        {
            return Err(ExecutionManagerError::Internal(format!(
                "runtime manager for {} is in unexpected state {state:?}",
                record.id
            )));
        }
        if matches!(
            managed_state(record)?,
            ManagedExecutionState::Starting | ManagedExecutionState::RestartStarting
        ) && !exec_endpoint_ready(manager.exec_socket_path()).await
        {
            return Ok(LocalExecutionObservation {
                state: ExecutionState::Creating,
                handle: None,
                exit_code: None,
            });
        }
        let visible_state = visible_active_state(record)?;
        let handle = match self.handle_from_manager(record, &manager).await {
            Ok(handle) => handle,
            Err(ExecutionManagerError::NotFound(_)) => {
                // The runtime state and health probes can still report a live
                // Sandbox after its host process has exited. Treat the missing
                // handle as the same terminal signal as a failed health probe
                // and wait for the exact status before projecting provider
                // loss. This closes the restart/replay race without creating a
                // second lifecycle path.
                return self
                    .finish_registered_terminal(record, &shared, manager, preserve_rootfs, None)
                    .await;
            }
            Err(error) => return Err(error),
        };
        Ok(LocalExecutionObservation {
            state: visible_state,
            handle: Some(handle),
            exit_code: None,
        })
    }

    async fn finish_registered_terminal(
        &self,
        record: &BoxRecord,
        shared: &SharedVm,
        mut manager: tokio::sync::MutexGuard<'_, VmManager>,
        preserve_rootfs: bool,
        exit_code: Option<i32>,
    ) -> ExecutionManagerResult<LocalExecutionObservation> {
        // A workload can exit between the first non-blocking wait and the
        // following health probe. A3S OCI publishes the exact status shortly
        // after the process becomes non-running, so retain its terminal record
        // and poll within a strict bound before any teardown.
        let deadline = tokio::time::Instant::now() + TERMINAL_EXIT_POLL_TIMEOUT;
        let mut exit_code = manager.exit_code().or(exit_code);
        while exit_code.is_none() {
            exit_code = manager
                .try_wait_exit()
                .await
                .map_err(|error| runtime_error("collect exit status", record, error))?;
            if exit_code.is_some() || tokio::time::Instant::now() >= deadline {
                break;
            }
            tokio::time::sleep(TERMINAL_EXIT_POLL_INTERVAL).await;
        }
        let exit_code = exit_code.ok_or_else(|| {
            ExecutionManagerError::Unavailable(format!(
                "runtime reported execution {} as terminal before its exact exit status became available",
                record.id
            ))
        })?;
        let cleanup = destroy_after_observation(&mut manager, preserve_rootfs).await;
        drop(manager);
        self.remove_manager(&record.id, shared);
        cleanup.map_err(|error| runtime_error("clean up", record, error))?;
        Ok(LocalExecutionObservation {
            state: ExecutionState::Stopped,
            handle: None,
            exit_code: Some(exit_code),
        })
    }

    async fn promote_if_ready(&self, record: &BoxRecord, manager: &mut VmManager) -> bool {
        let socket_dir = crate::vm::runtime_socket_dir(&self.home_dir, &record.id);
        let exec_socket = socket_dir.join("exec.sock");
        if !exec_endpoint_ready(Some(&exec_socket)).await {
            return false;
        }
        manager.exec_socket_path = Some(exec_socket);
        manager.pty_socket_path = Some(socket_dir.join("pty.sock"));
        manager.port_forward_socket_path = Some(socket_dir.join("portfwd.sock"));
        *manager.state.write().await = crate::BoxState::Ready;
        true
    }

    async fn recover_microvm(&self, record: &BoxRecord) -> ExecutionManagerResult<SharedVm> {
        self.metadata(record)?;
        let execution_id = execution_id(record)?;
        let execution_id_label = record.id.clone();
        let recorded = record.pid.map(|pid| (pid, record.pid_start_time));
        let located = tokio::task::spawn_blocking(move || {
            locate_microvm_process(&execution_id_label, recorded)
        })
        .await
        .map_err(|error| {
            ExecutionManagerError::Internal(format!(
                "MicroVM process discovery task failed for {}: {error}",
                record.id
            ))
        })?
        .map_err(ExecutionManagerError::Internal)?
        .ok_or(ExecutionManagerError::NotFound(execution_id))?;
        self.attach_microvm(record, located).await
    }

    async fn attach_microvm(
        &self,
        record: &BoxRecord,
        located: LocatedProcess,
    ) -> ExecutionManagerResult<SharedVm> {
        let mut manager = self.new_manager(record)?;
        let socket_dir = crate::vm::runtime_socket_dir(&self.home_dir, &record.id);
        manager
            .attach_running_process(
                located.pid,
                socket_dir.join("exec.sock"),
                Some(socket_dir.join("pty.sock")),
            )
            .await
            .map_err(|error| runtime_error("recover", record, error))?;
        if located.start_time.is_some()
            && crate::process::pid_start_time(located.pid) != located.start_time
        {
            return Err(ExecutionManagerError::NotFound(execution_id(record)?));
        }
        let recovered = Arc::new(Mutex::new(manager));
        match self.managers.entry(record.id.clone()) {
            Entry::Occupied(entry) => Ok(Arc::clone(entry.get())),
            Entry::Vacant(entry) => {
                entry.insert(Arc::clone(&recovered));
                Ok(recovered)
            }
        }
    }

    #[cfg(not(windows))]
    async fn require_microvm(&self, record: &BoxRecord) -> ExecutionManagerResult<SharedVm> {
        match self.manager(&record.id) {
            Some(manager) => Ok(manager),
            None => self.recover_microvm(record).await,
        }
    }

    async fn destroy_registered(
        &self,
        record: &BoxRecord,
        shared: SharedVm,
        remove_anonymous_volumes: bool,
        force_preserve_rootfs: bool,
        timeout_secs: Option<u64>,
    ) -> ExecutionManagerResult<LocalExecutionTermination> {
        let mut manager = shared.lock().await;
        let mut anonymous_volumes = if manager.anonymous_volumes().is_empty() {
            record.anonymous_volumes.clone()
        } else {
            manager.anonymous_volumes().to_vec()
        };
        let result = match (
            graceful_stop_options(record, timeout_secs)?,
            force_preserve_rootfs,
        ) {
            (Some((signal, timeout_ms)), true) => {
                manager
                    .destroy_preserving_rootfs_with_options(signal, timeout_ms)
                    .await
            }
            (Some((signal, timeout_ms)), false) => {
                manager.destroy_with_options(signal, timeout_ms).await
            }
            (None, true) => manager.destroy_preserving_rootfs().await,
            (None, false) => manager.destroy().await,
        };
        let exit_code = manager.exit_code();
        drop(manager);
        self.remove_manager(&record.id, &shared);
        result.map_err(|error| runtime_error("kill", record, error))?;
        if remove_anonymous_volumes {
            if anonymous_volumes.is_empty() {
                anonymous_volumes = self.anonymous_volumes_for_record(record).await;
            }
            self.cleanup_anonymous_volumes(&record.id, anonymous_volumes)
                .await;
        }
        Ok(LocalExecutionTermination {
            outcome: KillOutcome::Killed,
            exit_code,
        })
    }

    async fn anonymous_volumes_for_record(&self, record: &BoxRecord) -> Vec<String> {
        if !record.anonymous_volumes.is_empty() {
            return record.anonymous_volumes.clone();
        }
        let home_dir = self.home_dir.clone();
        let execution_id = record.id.clone();
        let short_id = record.id.chars().take(8).collect::<String>();
        let result = tokio::task::spawn_blocking(move || -> a3s_box_core::Result<Vec<String>> {
            let store =
                crate::VolumeStore::new(home_dir.join("volumes.json"), home_dir.join("volumes"));
            let prefix = format!("anon_{short_id}_");
            let mut names = store
                .load()?
                .into_values()
                .filter(|volume| {
                    volume
                        .labels
                        .get("anonymous")
                        .is_some_and(|value| value == "true")
                        && (volume.in_use_by.iter().any(|id| id == &execution_id)
                            || volume.name.starts_with(&prefix))
                })
                .map(|volume| volume.name)
                .collect::<Vec<_>>();
            names.sort();
            Ok(names)
        })
        .await;
        match result {
            Ok(Ok(names)) => names,
            Ok(Err(error)) => {
                tracing::warn!(
                    execution_id = %record.id,
                    %error,
                    "Failed to load anonymous volumes during managed cleanup"
                );
                Vec::new()
            }
            Err(error) => {
                tracing::warn!(
                    execution_id = %record.id,
                    %error,
                    "Anonymous volume recovery task failed"
                );
                Vec::new()
            }
        }
    }

    async fn cleanup_anonymous_volumes(&self, owner: &str, names: Vec<String>) {
        if names.is_empty() {
            return;
        }
        let home_dir = self.home_dir.clone();
        let owner = owner.to_string();
        let task = tokio::task::spawn_blocking(move || {
            let store = crate::VolumeStore::new(
                home_dir.join("volumes.json"),
                home_dir.join("volumes"),
            );
            for name in names {
                if let Err(error) = store.remove_anonymous(&name, &owner) {
                    tracing::warn!(volume = %name, %error, "Failed to remove managed anonymous volume");
                }
            }
        })
        .await;
        if let Err(error) = task {
            tracing::warn!(%error, "Anonymous volume cleanup task failed");
        }
    }

    async fn terminate_execution(
        &self,
        record: &BoxRecord,
    ) -> ExecutionManagerResult<LocalExecutionTermination> {
        let metadata = self.metadata(record)?;
        let remove_anonymous_volumes = record.auto_remove;
        let timeout_secs = record.stop_timeout;
        if let Some(manager) = self.manager(&record.id) {
            return self
                .destroy_registered(
                    record,
                    manager,
                    remove_anonymous_volumes,
                    false,
                    timeout_secs,
                )
                .await;
        }
        // A filesystem-only pause deliberately has no live provider evidence,
        // but a later terminal kill must still apply the configured rootfs and
        // anonymous-volume cleanup policy to the retained generation.
        if !metadata.paused_with_memory {
            let manager = Arc::new(Mutex::new(self.new_manager(record)?));
            return self
                .destroy_registered(
                    record,
                    manager,
                    remove_anonymous_volumes,
                    false,
                    timeout_secs,
                )
                .await;
        }
        match metadata.plan.backend {
            ExecutionBackend::A3sOci => {
                self.destroy_detached_sandbox(record, remove_anonymous_volumes, false, timeout_secs)
                    .await
            }
            ExecutionBackend::Krun => {
                let manager = self.recover_microvm(record).await?;
                self.destroy_registered(
                    record,
                    manager,
                    remove_anonymous_volumes,
                    false,
                    timeout_secs,
                )
                .await
            }
        }
    }
}

async fn destroy_after_observation(
    manager: &mut VmManager,
    preserve_rootfs: bool,
) -> a3s_box_core::Result<()> {
    if preserve_rootfs {
        manager.destroy_preserving_rootfs().await
    } else {
        manager.destroy().await
    }
}

#[async_trait]
impl LocalExecutionBackend for VmLocalExecutionBackend {
    fn route_for_create(
        &self,
        _record: &BoxRecord,
    ) -> ExecutionManagerResult<crate::ManagedRuntimeRoute> {
        Ok(crate::ManagedRuntimeRoute::BoxVm)
    }

    async fn start(&self, record: &BoxRecord) -> ExecutionManagerResult<LocalExecutionHandle> {
        super::record::validate_record_health(record)?;
        self.metadata(record)?;
        let box_dir = record.box_dir.clone();
        let execution_id = record.id.clone();
        tokio::task::spawn_blocking(move || {
            crate::rootfs::stage_box_terminal_rootfs_metadata(&box_dir)
        })
        .await
        .map_err(|error| {
            ExecutionManagerError::Internal(format!(
                "rootfs metadata staging task failed for {execution_id}: {error}"
            ))
        })?
        .map_err(|error| {
            ExecutionManagerError::Internal(format!(
                "failed to stage rootfs metadata for {execution_id}: {error}"
            ))
        })?;
        let mut manager = self.new_manager(record)?;
        let requested_persistence = manager.config.persistent;
        if should_reuse_preserved_rootfs(record)?
            && crate::vm::persistent_rootfs_generation_exists(&record.box_dir)
                .map_err(|error| runtime_error("inspect retained rootfs", record, error))?
        {
            manager.config.persistent = true;
        }
        let manager = Arc::new(Mutex::new(manager));
        match self.managers.entry(record.id.clone()) {
            Entry::Occupied(_) => {
                return Err(ExecutionManagerError::Unavailable(format!(
                    "execution {} already has an in-process runtime owner",
                    record.id
                )))
            }
            Entry::Vacant(entry) => {
                entry.insert(Arc::clone(&manager));
            }
        }

        let mut guard = manager.lock().await;
        let resource_home = self.home_dir.clone();
        let resource_record = record.clone();
        let resources = match tokio::task::spawn_blocking(move || {
            ExecutionResourceGuard::prepare(&resource_home, &resource_record)
        })
        .await
        {
            Ok(Ok(resources)) => resources,
            Ok(Err(error)) => {
                drop(guard);
                self.remove_manager(&record.id, &manager);
                return Err(error);
            }
            Err(error) => {
                drop(guard);
                self.remove_manager(&record.id, &manager);
                return Err(ExecutionManagerError::Internal(format!(
                    "managed resource preparation task failed for {}: {error}",
                    record.id
                )));
            }
        };
        #[cfg(target_os = "linux")]
        self.claim_transient_registry_auth_for_boot(&mut guard);
        if let Err(error) = guard.boot().await {
            guard.config.persistent = requested_persistence;
            // Sandbox boot cleanup synchronously asks the authoritative OCI
            // handler for its exit status before tearing down transient host
            // artifacts. A very short task can therefore complete while boot
            // is still establishing readiness. Keep the manager whenever that
            // cleanup captured an exact status so the normal inspect path can
            // project it into the durable managed record.
            if guard.exit_code().is_some() {
                tracing::debug!(
                    execution_id = %record.id,
                    %error,
                    "Runtime completed while startup was establishing readiness"
                );
                resources.disarm();
                return Err(ExecutionManagerError::Unavailable(format!(
                    "execution {} completed during startup",
                    record.id
                )));
            }
            drop(guard);
            self.remove_manager(&record.id, &manager);
            let rollback = tokio::task::spawn_blocking(move || resources.rollback()).await;
            if let Err(rollback_error) = rollback {
                tracing::warn!(
                    execution_id = %record.id,
                    %rollback_error,
                    "Managed resource rollback task failed"
                );
            }
            return Err(runtime_error("start", record, error));
        }
        guard.config.persistent = requested_persistence;
        resources.disarm();
        let exited_during_start = guard
            .try_wait_exit()
            .await
            .map_err(|error| runtime_error("collect startup exit status", record, error))?
            .is_some()
            || guard.has_exited().await;
        if exited_during_start {
            return Err(ExecutionManagerError::Unavailable(format!(
                "execution {} completed during startup",
                record.id
            )));
        }
        self.handle_from_manager(record, &guard).await
    }

    async fn inspect(
        &self,
        record: &BoxRecord,
    ) -> ExecutionManagerResult<LocalExecutionObservation> {
        let metadata = self.metadata(record)?;
        if metadata.plan.backend.is_sandbox() {
            return self.inspect_sandbox(record).await;
        }
        if let Some(manager) = self.manager(&record.id) {
            return self.inspect_registered(record, manager).await;
        }
        let manager = self.recover_microvm(record).await?;
        self.inspect_registered(record, manager).await
    }

    async fn pause(
        &self,
        record: &BoxRecord,
        keep_memory: bool,
    ) -> ExecutionManagerResult<LocalExecutionHandle> {
        let metadata = self.metadata(record)?;
        if metadata.plan.backend.is_sandbox() {
            if !keep_memory {
                return Err(unsupported(
                    record,
                    "pause without memory retention",
                    "the Sandbox backend",
                ));
            }
            return self.pause_sandbox(record).await;
        }
        if !keep_memory {
            return Err(unsupported(
                record,
                "pause without memory retention",
                "the local MicroVM backend",
            ));
        }
        #[cfg(windows)]
        {
            Err(unsupported(
                record,
                "pause",
                "the local MicroVM backend on Windows",
            ))
        }
        #[cfg(not(windows))]
        {
            let shared = self.require_microvm(record).await?;
            let manager = shared.lock().await;
            require_recorded_pid(record, &manager).await?;
            manager
                .pause()
                .await
                .map_err(|error| runtime_error("pause", record, error))?;
            self.handle_from_manager(record, &manager).await
        }
    }

    async fn resume(&self, record: &BoxRecord) -> ExecutionManagerResult<LocalExecutionHandle> {
        let metadata = self.metadata(record)?;
        if metadata.plan.backend.is_sandbox() {
            return self.resume_sandbox(record).await;
        }
        #[cfg(windows)]
        {
            Err(unsupported(
                record,
                "resume",
                "the local MicroVM backend on Windows",
            ))
        }
        #[cfg(not(windows))]
        {
            let shared = self.require_microvm(record).await?;
            let manager = shared.lock().await;
            require_recorded_pid(record, &manager).await?;
            manager
                .resume()
                .await
                .map_err(|error| runtime_error("resume", record, error))?;
            self.handle_from_manager(record, &manager).await
        }
    }

    async fn prepare_quiescent_rootfs(&self, record: &BoxRecord) -> ExecutionManagerResult<()> {
        self.new_manager(record)?
            .prepare_preserved_rootfs()
            .map(|_| ())
            .map_err(|error| runtime_error("prepare quiescent rootfs", record, error))
    }

    async fn cleanup_quiescent_rootfs(&self, record: &BoxRecord) -> ExecutionManagerResult<()> {
        self.new_manager(record)?
            .cleanup_preserved_rootfs()
            .map_err(|error| runtime_error("clean up quiescent rootfs", record, error))
    }

    async fn kill(&self, record: &BoxRecord) -> ExecutionManagerResult<KillOutcome> {
        Ok(self.terminate_execution(record).await?.outcome)
    }

    async fn kill_with_status(
        &self,
        record: &BoxRecord,
    ) -> ExecutionManagerResult<LocalExecutionTermination> {
        self.terminate_execution(record).await
    }

    async fn stop_for_restart(
        &self,
        record: &BoxRecord,
        timeout_secs: Option<u64>,
    ) -> ExecutionManagerResult<KillOutcome> {
        let metadata = self.metadata(record)?;
        #[cfg(target_os = "linux")]
        if metadata.plan.backend.is_sandbox() {
            super::snapshot::persist_sandbox_snapshot_mappings(record)?;
        }
        let timeout_secs = timeout_secs.or(record.stop_timeout);
        if let Some(manager) = self.manager(&record.id) {
            return Ok(self
                .destroy_registered(record, manager, false, true, timeout_secs)
                .await?
                .outcome);
        }
        match metadata.plan.backend {
            ExecutionBackend::A3sOci => Ok(self
                .destroy_detached_sandbox(record, false, true, timeout_secs)
                .await?
                .outcome),
            ExecutionBackend::Krun => {
                let manager = self.recover_microvm(record).await?;
                Ok(self
                    .destroy_registered(record, manager, false, true, timeout_secs)
                    .await?
                    .outcome)
            }
        }
    }
}

pub(super) fn should_force_rootfs_preservation(record: &BoxRecord) -> ExecutionManagerResult<bool> {
    let state = super::support::managed_state(record)?;
    let metadata = record.managed_execution.as_ref().ok_or_else(|| {
        ExecutionManagerError::Internal(format!(
            "execution {} has no managed lifecycle metadata",
            record.id
        ))
    })?;
    Ok(match state {
        // A foreground one-shot can finish before readiness is published. Its
        // exact status is already terminal, but the caller still has to drain
        // stdout/stderr and archive an auto-removed result. Tear down mounts
        // and runtime processes while retaining the box directory until that
        // caller completes its normal terminal cleanup.
        ManagedExecutionState::Starting => true,
        ManagedExecutionState::Pausing => matches!(
            metadata.pending_operation.as_ref(),
            Some(ManagedExecutionOperation::Pause {
                keep_memory: false,
                ..
            })
        ),
        ManagedExecutionState::Resuming => !metadata.paused_with_memory,
        ManagedExecutionState::RestartStopping | ManagedExecutionState::RestartStarting => true,
        _ => false,
    })
}

fn should_reuse_preserved_rootfs(record: &BoxRecord) -> ExecutionManagerResult<bool> {
    Ok(matches!(
        super::support::managed_state(record)?,
        ManagedExecutionState::Resuming | ManagedExecutionState::RestartStarting
    ) && should_force_rootfs_preservation(record)?)
}

fn graceful_stop_options(
    record: &BoxRecord,
    timeout_secs: Option<u64>,
) -> ExecutionManagerResult<Option<(i32, u64)>> {
    if timeout_secs.is_none() && record.stop_signal.is_none() {
        return Ok(None);
    }
    let timeout_ms = timeout_secs
        .unwrap_or(DEFAULT_SHUTDOWN_TIMEOUT_MS / 1_000)
        .checked_mul(1_000)
        .ok_or_else(|| {
            ExecutionManagerError::InvalidRequest(format!(
                "stop timeout is too large for execution {}",
                record.id
            ))
        })?;
    let signal = record
        .stop_signal
        .as_deref()
        .map(a3s_box_core::vmm::parse_signal_name)
        .unwrap_or(libc::SIGTERM);
    Ok(Some((signal, timeout_ms)))
}

#[cfg(not(windows))]
async fn require_recorded_pid(
    record: &BoxRecord,
    manager: &VmManager,
) -> ExecutionManagerResult<()> {
    let execution_id = execution_id(record)?;
    let pid = manager
        .pid()
        .await
        .ok_or_else(|| ExecutionManagerError::NotFound(execution_id.clone()))?;
    if record.pid != Some(pid)
        || !crate::process::is_process_alive_with_identity(pid, record.pid_start_time)
    {
        return Err(ExecutionManagerError::NotFound(execution_id));
    }
    Ok(())
}

fn visible_active_state(record: &BoxRecord) -> ExecutionManagerResult<ExecutionState> {
    match managed_state(record)? {
        ManagedExecutionState::Paused => Ok(ExecutionState::Paused),
        ManagedExecutionState::Resuming => {
            let metadata = record.managed_execution.as_ref().ok_or_else(|| {
                ExecutionManagerError::Internal(format!(
                    "execution {} has no managed lifecycle metadata",
                    record.id
                ))
            })?;
            if metadata.paused_with_memory {
                Ok(ExecutionState::Paused)
            } else {
                // A cold-paused execution has no provider process. Any live
                // process observed while its resume is pending is therefore
                // the replacement generation started before record commit.
                Ok(ExecutionState::Running)
            }
        }
        ManagedExecutionState::Starting
        | ManagedExecutionState::RestartStarting
        | ManagedExecutionState::Running
        | ManagedExecutionState::Pausing
        | ManagedExecutionState::Killing => Ok(ExecutionState::Running),
        ManagedExecutionState::Snapshotting => match record
            .managed_execution
            .as_ref()
            .and_then(|metadata| metadata.pending_operation.as_ref())
        {
            Some(ManagedExecutionOperation::Snapshot {
                source_state: ManagedExecutionState::Running,
                ..
            }) => Ok(ExecutionState::Running),
            Some(ManagedExecutionOperation::Snapshot {
                source_state: ManagedExecutionState::Paused,
                ..
            }) => Ok(ExecutionState::Paused),
            _ => Err(ExecutionManagerError::Internal(format!(
                "execution {} has invalid snapshot metadata",
                record.id
            ))),
        },
        ManagedExecutionState::RestartStopping => match record
            .managed_execution
            .as_ref()
            .and_then(|metadata| metadata.pending_operation.as_ref())
        {
            Some(ManagedExecutionOperation::Restart {
                source_state: ManagedExecutionState::Paused,
                ..
            }) => Ok(ExecutionState::Paused),
            Some(ManagedExecutionOperation::Restart {
                source_state: ManagedExecutionState::Running,
                ..
            }) => Ok(ExecutionState::Running),
            _ => Err(ExecutionManagerError::Internal(format!(
                "execution {} has invalid restart teardown metadata",
                record.id
            ))),
        },
        state => Err(ExecutionManagerError::Internal(format!(
            "execution {} has no active runtime in managed state {state}",
            record.id
        ))),
    }
}

fn managed_state(record: &BoxRecord) -> ExecutionManagerResult<ManagedExecutionState> {
    record
        .managed_state()
        .map_err(|error| ExecutionManagerError::Internal(error.to_string()))?
        .ok_or_else(|| {
            ExecutionManagerError::Internal(format!("execution {} is not managed", record.id))
        })
}

fn execution_id(record: &BoxRecord) -> ExecutionManagerResult<ExecutionId> {
    ExecutionId::new(record.id.clone())
        .map_err(|error| ExecutionManagerError::Internal(error.to_string()))
}

fn runtime_error(
    action: &str,
    record: &BoxRecord,
    error: impl std::fmt::Display,
) -> ExecutionManagerError {
    ExecutionManagerError::Internal(format!(
        "failed to {action} execution {}: {error}",
        record.id
    ))
}

fn unsupported(record: &BoxRecord, operation: &str, backend: &str) -> ExecutionManagerError {
    ExecutionManagerError::Unavailable(format!(
        "{operation} is not supported by {backend} for execution {}",
        record.id
    ))
}

#[cfg(unix)]
async fn exec_endpoint_ready(path: Option<&Path>) -> bool {
    let Some(path) = path else {
        return false;
    };
    let attempt = async {
        let client = crate::ExecClient::connect(path).await.ok()?;
        client.heartbeat().await.ok().filter(|ready| *ready)
    };
    tokio::time::timeout(Duration::from_millis(500), attempt)
        .await
        .ok()
        .flatten()
        .is_some()
}

#[cfg(not(unix))]
async fn exec_endpoint_ready(path: Option<&Path>) -> bool {
    path.is_some()
}

#[cfg(test)]
#[path = "vm_backend_tests.rs"]
mod tests;