runtime: separate provider control from sandbox lifecycle

crates/libs/node/mgr/server/src/service.rs

@@ -743,3 +743,5 @@ pub(crate) use cleanup_helpers::neutralize_retained_cache_owned_by_uid;
 use network_helpers::*;
 
 use crate::snapshot_request::{FirecrackerCapture, FirecrackerRestore};
+
+mod runtime_control;

crates/libs/node/mgr/server/src/service/launch_prepare.rs

@@ -305,15 +305,15 @@ impl NodeManagerService {
             .await?;
         let firecracker_bin = resolved_execution
             .runtime_files
-            .path(RuntimeFileRole::FirecrackerBinary)
+            .path(RuntimeFileRole::RuntimeExecutable)
             .ok_or_else(|| {
                 Status::failed_precondition("resolved Firecracker runtime is missing firecracker")
             })?
             .display()
             .to_string();
         let jailer_bin = resolved_execution
             .runtime_files
-            .path(RuntimeFileRole::JailerBinary)
+            .path(RuntimeFileRole::LauncherExecutable)
             .ok_or_else(|| {
                 Status::failed_precondition("resolved Firecracker runtime is missing jailer")
             })?

crates/libs/node/mgr/server/src/service/runtime_control.rs

@@ -0,0 +1,50 @@
+//! Derive runtime control from recorded execution, never a caller-selected socket.
+use super::validation::validate_sandbox_id;
+use super::*;
+use pplx_space_common_core::types::{ExecutionMetadata, SandboxArtifactMetadata};
+use std::io::Read;
+
+impl NodeManagerService {
+    pub(super) async fn recorded_execution(
+        &self,
+        sandbox_id: String,
+    ) -> Result<Option<ExecutionMetadata>, Status> {
+        validate_sandbox_id(&sandbox_id)?;
+        self.off_runtime("read sandbox execution", move || {
+            let directory_path = PathBuf::from(format!("{CACHE_DIR}/{sandbox_id}"));
+            match std::fs::symlink_metadata(directory_path.join("metadata.json")) {
+                Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
+                Err(error) => {
+                    return Err(Status::failed_precondition(format!(
+                        "inspect sandbox execution: {error}"
+                    )));
+                }
+                Ok(_) => {}
+            }
+            let directory = trusted_paths::open_directory(&directory_path)?;
+            let file = trusted_paths::open_regular_file(&directory, "metadata.json".as_ref())?;
+            let mut bytes = Vec::new();
+            file.take(1024 * 1024 + 1)
+                .read_to_end(&mut bytes)
+                .map_err(|error| {
+                    Status::failed_precondition(format!("read sandbox execution: {error}"))
+                })?;
+            if bytes.len() > 1024 * 1024 {
+                return Err(Status::failed_precondition(
+                    "sandbox execution metadata exceeds 1 MiB",
+                ));
+            }
+            let metadata: SandboxArtifactMetadata =
+                serde_json::from_slice(&bytes).map_err(|error| {
+                    Status::failed_precondition(format!("invalid sandbox execution: {error}"))
+                })?;
+            if metadata.sandbox_id.to_string() != sandbox_id {
+                return Err(Status::failed_precondition(
+                    "sandbox execution metadata identity mismatch",
+                ));
+            }
+            Ok(Some(metadata.execution))
+        })
+        .await
+    }
+}

crates/libs/node/mgr/server/src/service/snapshots.rs

@@ -1,4 +1,4 @@
 mod environment;
 
 use super::cleanup_helpers::*;
 use super::disk::*;
@@ -206,26 +206,27 @@
             .sandbox_id
             .parse::<SandboxId>()
             .map_err(|error| Status::invalid_argument(error.to_string()))?;
-        let api_socket = format!("{JAIL_DIR}/firecracker/{sandbox_id}/root/firecracker.sock");
-        let state = tokio::time::timeout(
-            FIRECRACKER_VM_STATE_TIMEOUT,
-            crate::fc_api::FcApiClient::new(api_socket).get_vm_state(),
-        )
-        .await
-        .map_err(|_| Status::deadline_exceeded("Firecracker VM-state probe timed out"))?
-        .map_err(|error| {
-            Status::unavailable(format!("Firecracker VM-state probe failed: {error}"))
-        })?;
+        let execution = self.recorded_execution(sandbox_id.to_string()).await?;
+        // Pre-metadata historical sandboxes are Firecracker. New adapters must
+        // persist their execution before launch; missing evidence cannot select CH.
+        let runtime = execution.as_ref().map_or(
+            pplx_space_common_core::types::RuntimeKind::Firecracker,
+            |execution| execution.runtime().kind,
+        );
+        let socket = pplx_space_node_runtime_core::control::socket_path(runtime, sandbox_id)
+            .map_err(|error| Status::failed_precondition(error.to_string()))?;
+        let control = pplx_space_node_runtime_core::control::connect(runtime, socket)
+            .map_err(|error| Status::failed_precondition(error.to_string()))?;
+        let state = tokio::time::timeout(FIRECRACKER_VM_STATE_TIMEOUT, control.state())
+            .await
+            .map_err(|_| Status::deadline_exceeded("runtime state probe timed out"))?
+            .map_err(|error| Status::unavailable(format!("runtime state probe failed: {error}")))?;
+        use pplx_space_node_runtime_core::vm_state::VmState;
         let state = match state {
-            pplx_space_node_runtime_core::fc_api::FirecrackerVmState::NotStarted => {
-                crate::proto::VmState::NotStarted
-            }
-            pplx_space_node_runtime_core::fc_api::FirecrackerVmState::Running => {
-                crate::proto::VmState::Running
-            }
-            pplx_space_node_runtime_core::fc_api::FirecrackerVmState::Paused => {
-                crate::proto::VmState::Paused
-            }
+            VmState::NotStarted => crate::proto::VmState::NotStarted,
+            VmState::Running => crate::proto::VmState::Running,
+            VmState::Paused => crate::proto::VmState::Paused,
+            VmState::Shutdown => crate::proto::VmState::Shutdown,
         };
         Ok(Response::new(GetVmStateResponse {
             state: state.into(),

crates/libs/node/mgr/server/src/service/snapshots/environment.rs

@@ -1,9 +1,8 @@
 //! Capture admission keeps an environment's restore requirement on the node.
 
 use super::super::*;
-use pplx_space_common_core::types::{EnvironmentReference, SandboxArtifactMetadata};
+use pplx_space_common_core::types::EnvironmentReference;
 use pplx_space_node_runtime_core::runtime_inventory::ExecutionSelectionMode;
-use std::io::Read;
 
 use super::super::node_artifacts::PolicyScope;
 
@@ -39,43 +38,7 @@ impl NodeManagerService {
                 ExecutionSelectionMode::PersistedStateRestore,
             )?;
         }
-        let execution = self
-            .off_runtime("inspect snapshot execution", move || {
-                let directory_path = PathBuf::from(format!("{CACHE_DIR}/{sandbox_id}"));
-                match std::fs::symlink_metadata(directory_path.join("metadata.json")) {
-                    Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
-                    Err(error) => {
-                        return Err(Status::failed_precondition(format!(
-                            "inspect snapshot execution: {error}"
-                        )));
-                    }
-                    Ok(_) => {}
-                }
-                let directory = trusted_paths::open_directory(&directory_path)?;
-                let file = trusted_paths::open_regular_file(&directory, "metadata.json".as_ref())?;
-                let mut bytes = Vec::new();
-                file.take(1024 * 1024 + 1)
-                    .read_to_end(&mut bytes)
-                    .map_err(|error| {
-                        Status::failed_precondition(format!("read snapshot execution: {error}"))
-                    })?;
-                if bytes.len() > 1024 * 1024 {
-                    return Err(Status::failed_precondition(
-                        "snapshot execution metadata exceeds 1 MiB",
-                    ));
-                }
-                let metadata: SandboxArtifactMetadata =
-                    serde_json::from_slice(&bytes).map_err(|error| {
-                        Status::failed_precondition(format!("invalid snapshot execution: {error}"))
-                    })?;
-                if metadata.sandbox_id.to_string() != sandbox_id {
-                    return Err(Status::failed_precondition(
-                        "snapshot environment differs from the running sandbox",
-                    ));
-                }
-                Ok(Some(metadata.execution))
-            })
-            .await?;
+        let execution = self.recorded_execution(sandbox_id.clone()).await?;
         let Some(execution) = execution else {
             return if reference.is_some() {
                 Err(Status::failed_precondition(

crates/libs/node/runtime/core/src/control.rs

@@ -0,0 +1,85 @@
+//! Runtime control owned by node-manager. Lifecycle services do not receive a
+//! VMM's host-path-capable control socket or interpret provider-specific states.
+use pplx_space_common_core::types::RuntimeKind;
+use std::{future::Future, path::PathBuf, pin::Pin};
+
+mod cloud_hypervisor;
+
+pub use crate::vm_state::VmState;
+
+#[derive(Debug, thiserror::Error)]
+pub enum ControlError {
+    #[error("runtime {0} has no control adapter")]
+    UnsupportedRuntime(RuntimeKind),
+    #[error("{operation} is not supported by runtime {runtime}")]
+    UnsupportedOperation {
+        runtime: RuntimeKind,
+        operation: &'static str,
+    },
+    #[error(transparent)]
+    Firecracker(#[from] crate::fc_api::FcApiError),
+    #[error(transparent)]
+    CloudHypervisor(#[from] cloud_hypervisor::ApiError),
+}
+
+pub type ControlFuture<'a, T> = Pin<Box<dyn Future<Output = Result<T, ControlError>> + Send + 'a>>;
+
+/// Control is independent of process launch: an in-process runtime can implement
+/// this interface without pretending to expose an HTTP socket or jailer.
+pub trait RuntimeControl: Send + Sync {
+    fn state(&self) -> ControlFuture<'_, VmState>;
+    fn pause(&self) -> ControlFuture<'_, ()>;
+    fn resume(&self) -> ControlFuture<'_, ()>;
+}
+
+/// Construct only in the process authorized to own this runtime's control plane.
+/// The socket is a locally derived adapter locator, never a tenant input.
+pub fn connect(
+    runtime: RuntimeKind,
+    socket: PathBuf,
+) -> Result<Box<dyn RuntimeControl>, ControlError> {
+    match runtime {
+        RuntimeKind::Firecracker => Ok(Box::new(crate::fc_api::FcApiClient::new(socket))),
+        RuntimeKind::CloudHypervisor => Ok(Box::new(cloud_hypervisor::Client::new(socket))),
+        RuntimeKind::Gvisor => Err(ControlError::UnsupportedRuntime(runtime)),
+    }
+}
+
+impl RuntimeControl for crate::fc_api::FcApiClient {
+    fn state(&self) -> ControlFuture<'_, VmState> {
+        Box::pin(async move {
+            use crate::fc_api::FirecrackerVmState;
+            Ok(match self.get_vm_state().await? {
+                FirecrackerVmState::NotStarted => VmState::NotStarted,
+                FirecrackerVmState::Running => VmState::Running,
+                FirecrackerVmState::Paused => VmState::Paused,
+            })
+        })
+    }
+    fn pause(&self) -> ControlFuture<'_, ()> {
+        Box::pin(async move { Ok(crate::fc_api::FcApiClient::pause(self).await?) })
+    }
+    fn resume(&self) -> ControlFuture<'_, ()> {
+        Box::pin(async move { Ok(crate::fc_api::FcApiClient::resume(self).await?) })
+    }
+}
+
+/// Linux standalone adapters keep control endpoints in their own isolation root.
+/// The shared control trait does not require a socket-backed implementation.
+pub fn socket_path(
+    runtime: RuntimeKind,
+    sandbox_id: pplx_space_common_core::types::SandboxId,
+) -> Result<PathBuf, ControlError> {
+    let filename = match runtime {
+        RuntimeKind::Firecracker => "firecracker.sock",
+        RuntimeKind::CloudHypervisor => "control.sock",
+        RuntimeKind::Gvisor => return Err(ControlError::UnsupportedRuntime(runtime)),
+    };
+    Ok(
+        std::path::Path::new(pplx_space_common_core::host::linux::paths::FIRECRACKER_JAIL_DIR)
+            .join(runtime.as_str())
+            .join(sandbox_id.to_string())
+            .join("root")
+            .join(filename),
+    )
+}

crates/libs/node/runtime/core/src/control/cloud_hypervisor.rs

@@ -0,0 +1,125 @@
+use super::{ControlError, ControlFuture, RuntimeControl, VmState};
+use http_body_util::{BodyExt, Empty, Limited};
+use hyper::{Method, Request, StatusCode, body::Bytes};
+use hyper_util::rt::TokioIo;
+use pplx_space_common_core::types::RuntimeKind;
+use serde::Deserialize;
+use std::{path::PathBuf, time::Duration};
+
+const REQUEST_TIMEOUT: Duration = Duration::from_secs(2);
+const MAX_RESPONSE_BYTES: usize = 64 * 1024;
+
+#[derive(Debug, thiserror::Error)]
+pub enum ApiError {
+    #[error("Cloud Hypervisor API I/O: {0}")]
+    Io(#[from] std::io::Error),
+    #[error("Cloud Hypervisor API transport: {0}")]
+    Transport(#[from] hyper::Error),
+    #[error("Cloud Hypervisor API request: {0}")]
+    Request(#[from] hyper::http::Error),
+    #[error("Cloud Hypervisor API response exceeds its bound or cannot be read: {0}")]
+    Body(#[source] Box<dyn std::error::Error + Send + Sync>),
+    #[error("Cloud Hypervisor API returned HTTP {0}")]
+    Status(StatusCode),
+    #[error("Cloud Hypervisor API returned invalid VM information: {0}")]
+    Decode(#[from] serde_json::Error),
+    #[error("Cloud Hypervisor API deadline exceeded")]
+    Timeout,
+}
+
+pub(super) struct Client {
+    socket: PathBuf,
+}
+
+impl Client {
+    pub(super) fn new(socket: PathBuf) -> Self {
+        Self { socket }
+    }
+
+    /// One deadline covers connect, handshake, headers and a bounded body. A
+    /// dropped caller aborts the connection task; control requests are not replayed.
+    async fn info(&self) -> Result<Info, ApiError> {
+        tokio::time::timeout(REQUEST_TIMEOUT, async {
+            let stream = tokio::net::UnixStream::connect(&self.socket).await?;
+            let (mut sender, connection) =
+                hyper::client::conn::http1::handshake(TokioIo::new(stream)).await?;
+            let task = ConnectionTask(tokio::spawn(async move {
+                if let Err(error) = connection.await {
+                    tracing::debug!(%error, "runtime control connection closed");
+                }
+            }));
+            let request = Request::builder()
+                .method(Method::GET)
+                .uri("/api/v1/vm.info")
+                .header("Host", "localhost")
+                .body(Empty::<Bytes>::new())?;
+            let response = sender.send_request(request).await?;
+            if !response.status().is_success() {
+                return Err(ApiError::Status(response.status()));
+            }
+            let body = Limited::new(response.into_body(), MAX_RESPONSE_BYTES)
+                .collect()
+                .await
+                .map_err(ApiError::Body)?
+                .to_bytes();
+            let result = serde_json::from_slice(&body).map_err(ApiError::Decode);
+            drop(task);
+            result
+        })
+        .await
+        .map_err(|_| ApiError::Timeout)?
+    }
+}
+
+struct ConnectionTask(tokio::task::JoinHandle<()>);
+impl Drop for ConnectionTask {
+    fn drop(&mut self) {
+        self.0.abort();
+    }
+}
+
+#[derive(Deserialize)]
+struct Info {
+    state: State,
+}
+
+/// Upstream v54 VM states. Unknown values are an observation error, never a
+/// guessed lifecycle state. Other response fields are owned by the CH adapter.
+#[derive(Deserialize)]
+enum State {
+    Created,
+    Running,
+    Paused,
+    Shutdown,
+}
+
+impl RuntimeControl for Client {
+    fn state(&self) -> ControlFuture<'_, VmState> {
+        Box::pin(async move {
+            Ok(match self.info().await?.state {
+                State::Created => VmState::NotStarted,
+                State::Running => VmState::Running,
+                State::Paused => VmState::Paused,
+                State::Shutdown => VmState::Shutdown,
+            })
+        })
+    }
+    fn pause(&self) -> ControlFuture<'_, ()> {
+        unsupported("pause")
+    }
+    fn resume(&self) -> ControlFuture<'_, ()> {
+        unsupported("resume")
+    }
+}
+
+fn unsupported(operation: &'static str) -> ControlFuture<'static, ()> {
+    Box::pin(async move {
+        Err(ControlError::UnsupportedOperation {
+            runtime: RuntimeKind::CloudHypervisor,
+            operation,
+        })
+    })
+}
+
+#[cfg(test)]
+mod tests;