use std::path::{Path, PathBuf}; use clusterflux_core::{ Capability, CommandBackendKind, CommandInvocation, CommandPlan, Digest, EnvironmentKind, GuestRuntimeKind, ProcessId, TaskInstanceId, VfsObject, VfsOverlay, VfsPath, }; use serde::{Deserialize, Serialize}; use sha2::{Digest as ShaDigest, Sha256}; use thiserror::Error; mod command_runner; mod source_snapshot; mod windows_dev; pub use clusterflux_wasm_runtime::{ WasmDebugControl, WasmTaskError, WasmTaskHost, WasmtimeDebugProbe, WasmtimeTaskRuntime, }; use command_runner::capture_command_logs; pub use command_runner::{ authorize_node_command, CapturedCommandLogs, CommandOutput, LocalCommandExecutor, VirtualThreadCommand, DEFAULT_COMMAND_LOG_LIMIT_BYTES, }; pub use windows_dev::{WindowsCommandDevBackend, WindowsSandboxStubBackend}; pub fn snapshot_project_digest(project_root: &Path) -> Result { source_snapshot::snapshot_project(project_root).map(|snapshot| snapshot.digest) } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct MaterializedEnvironment { pub name: String, pub backend: CommandBackendKind, pub local_reference: String, } #[derive(Clone, Debug, Error, PartialEq, Eq)] pub enum BackendError { #[error("native command denied: {0}")] Denied(String), #[error("environment is required for this backend")] MissingEnvironment, #[error("command failed to execute: {0}")] Command(String), #[error("task execution cancelled: {0}")] Cancelled(String), #[error("artifact staging failed: {0}")] Artifact(String), #[error("unsupported environment kind for backend")] UnsupportedEnvironment, #[error("node cannot freeze task `{task}` for the current debug epoch")] DebugFreezeUnsupported { task: TaskInstanceId }, } pub trait CommandBackend { fn kind(&self) -> CommandBackendKind; fn plan(&self, invocation: &CommandInvocation) -> Result; } #[derive(Clone, Debug, Default)] pub struct LinuxRootlessPodmanBackend; fn podman_container_identity(process: &ProcessId, task: &TaskInstanceId) -> String { let mut digest = Sha256::new(); digest.update(process.to_string().as_bytes()); digest.update([0]); digest.update(task.to_string().as_bytes()); let suffix = digest .finalize() .iter() .take(12) .map(|byte| format!("{byte:02x}")) .collect::(); format!("clusterflux-{suffix}") } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct PodmanCommand { pub program: String, pub args: Vec, } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct ProcessOutput { pub status_code: Option, pub stdout: Vec, pub stderr: Vec, } pub trait ProcessRunner { fn run(&mut self, command: &PodmanCommand) -> Result; } #[derive(Clone, Debug, Default)] pub struct StdProcessRunner; impl ProcessRunner for StdProcessRunner { fn run(&mut self, command: &PodmanCommand) -> Result { let output = std::process::Command::new(&command.program) .args(&command.args) .output() .map_err(|err| BackendError::Command(format!("{err:#}")))?; Ok(ProcessOutput { status_code: output.status.code(), stdout: output.stdout, stderr: output.stderr, }) } } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct PodmanEnvironmentMaterialization { pub environment: String, pub image_tag: String, pub build: PodmanCommand, pub rootless_user_podman: bool, pub embeds_full_image_in_bundle: bool, } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct LocalSourceCheckout { pub host_path: PathBuf, pub snapshot: Digest, } #[derive(Clone, Debug)] pub struct LocalCheckoutTaskRequest<'a> { pub process: ProcessId, pub virtual_thread: TaskInstanceId, pub invocation: &'a CommandInvocation, pub checkout: LocalSourceCheckout, pub output_root: PathBuf, pub stage_stdout_as: Option, } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub enum SourceAccessMode { LocalCheckoutBindMount { host_path: PathBuf, container_path: String, read_only: bool, snapshot: Digest, }, NodePreparedSnapshot { node_path: PathBuf, container_path: String, snapshot: Digest, }, } impl SourceAccessMode { pub fn uses_full_repo_tarball(&self) -> bool { false } pub fn coordinator_routed_file_reads(&self) -> bool { false } } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub enum LinuxTaskState { Running, Frozen, Cancelled, Completed, } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct LinuxTaskLifecycle { pub process: ProcessId, pub virtual_thread: TaskInstanceId, pub log_stream: String, pub cancellation_token: String, pub debug_participant: String, pub freeze_supported: bool, pub state: LinuxTaskState, } impl LinuxTaskLifecycle { pub fn new(process: ProcessId, virtual_thread: TaskInstanceId) -> Self { Self { log_stream: format!("process/{process}/task/{virtual_thread}/logs"), cancellation_token: format!("cancel:{process}:{virtual_thread}"), debug_participant: format!("debug:{process}:{virtual_thread}"), process, virtual_thread, freeze_supported: true, state: LinuxTaskState::Running, } } pub fn freeze_for_debug_epoch(&mut self) -> Result<(), BackendError> { if !self.freeze_supported { return Err(BackendError::DebugFreezeUnsupported { task: self.virtual_thread.clone(), }); } self.state = LinuxTaskState::Frozen; Ok(()) } pub fn resume_after_debug_epoch(&mut self) { if self.state == LinuxTaskState::Frozen { self.state = LinuxTaskState::Running; } } pub fn cancel(&mut self) { self.state = LinuxTaskState::Cancelled; } pub fn complete(&mut self) { self.state = LinuxTaskState::Completed; } } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct LinuxCommandRunPlan { pub process: ProcessId, pub virtual_thread: TaskInstanceId, pub image_tag: String, pub run: PodmanCommand, pub source_access: SourceAccessMode, pub output_root: PathBuf, pub stage_stdout_as: Option, pub uses_full_repo_tarball: bool, pub coordinator_routed_file_reads: bool, pub lifecycle: LinuxTaskLifecycle, } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct LinuxCommandTaskOutput { pub virtual_thread: TaskInstanceId, pub status_code: Option, pub stdout: String, pub stderr: String, pub stdout_truncated: bool, pub stderr_truncated: bool, pub log_backpressured: bool, pub staged_artifact: Option, pub lifecycle: LinuxTaskLifecycle, } impl LinuxRootlessPodmanBackend { pub fn materialize_environment( &self, env: &clusterflux_core::EnvironmentResource, ) -> Result { match env.kind { EnvironmentKind::Containerfile | EnvironmentKind::Dockerfile => {} EnvironmentKind::NixFlake => return Err(BackendError::UnsupportedEnvironment), } let image_tag = self.image_tag(env); Ok(PodmanEnvironmentMaterialization { environment: env.name.clone(), image_tag: image_tag.clone(), build: PodmanCommand { program: "podman".to_owned(), args: vec![ "build".to_owned(), "--pull=missing".to_owned(), "--tag".to_owned(), image_tag, "--file".to_owned(), env.recipe_path.to_string_lossy().into_owned(), env.context_path.to_string_lossy().into_owned(), ], }, rootless_user_podman: true, embeds_full_image_in_bundle: false, }) } pub fn plan_local_checkout_run( &self, process: ProcessId, virtual_thread: TaskInstanceId, invocation: &CommandInvocation, checkout: LocalSourceCheckout, output_root: PathBuf, stage_stdout_as: Option, ) -> Result { let env = invocation .env .as_ref() .ok_or(BackendError::MissingEnvironment)?; let materialization = self.materialize_environment(env)?; let source_access = SourceAccessMode::LocalCheckoutBindMount { host_path: checkout.host_path.clone(), container_path: "/workspace".to_owned(), read_only: true, snapshot: checkout.snapshot, }; let lifecycle = LinuxTaskLifecycle::new(process.clone(), virtual_thread.clone()); let container_identity = podman_container_identity(&process, &virtual_thread); let network = match invocation.network { clusterflux_core::CommandNetworkPolicy::Disabled => "none", clusterflux_core::CommandNetworkPolicy::Enabled => "slirp4netns", }; let mut args = vec![ "run".to_owned(), "--rm".to_owned(), "--name".to_owned(), container_identity, "--network".to_owned(), network.to_owned(), "--cpus".to_owned(), "2".to_owned(), "--memory".to_owned(), "2g".to_owned(), "--pids-limit".to_owned(), "256".to_owned(), "--security-opt".to_owned(), "no-new-privileges".to_owned(), "--cap-drop".to_owned(), "all".to_owned(), "--volume".to_owned(), format!("{}:/workspace:ro,Z", checkout.host_path.to_string_lossy()), "--volume".to_owned(), format!("{}:/clusterflux/output:rw,Z", output_root.to_string_lossy()), "--env".to_owned(), "CARGO_TARGET_DIR=/clusterflux/output/target".to_owned(), "--workdir".to_owned(), invocation.working_directory.clone(), ]; for (name, value) in &invocation.environment_variables { args.push("--env".to_owned()); args.push(format!("{name}={value}")); } args.push(materialization.image_tag.clone()); args.push(invocation.program.clone()); args.extend(invocation.args.iter().cloned()); Ok(LinuxCommandRunPlan { process, virtual_thread, image_tag: materialization.image_tag, run: PodmanCommand { program: "podman".to_owned(), args, }, source_access, output_root, stage_stdout_as, uses_full_repo_tarball: false, coordinator_routed_file_reads: false, lifecycle, }) } pub fn execute_environment_materialization( &self, env: &clusterflux_core::EnvironmentResource, runner: &mut impl ProcessRunner, ) -> Result { let materialization = self.materialize_environment(env)?; let output = runner.run(&materialization.build)?; if output.status_code != Some(0) { return Err(BackendError::Command(format!( "podman build for environment `{}` failed with status {:?}: {}", materialization.environment, output.status_code, String::from_utf8_lossy(&output.stderr) ))); } Ok(MaterializedEnvironment { name: materialization.environment, backend: CommandBackendKind::LinuxRootlessPodman, local_reference: materialization.image_tag, }) } pub fn execute_run_plan( &self, plan: LinuxCommandRunPlan, runner: &mut impl ProcessRunner, overlay: &mut VfsOverlay, ) -> Result { self.execute_run_plan_with_log_limit(plan, runner, overlay, DEFAULT_COMMAND_LOG_LIMIT_BYTES) } pub fn execute_run_plan_with_log_limit( &self, mut plan: LinuxCommandRunPlan, runner: &mut impl ProcessRunner, overlay: &mut VfsOverlay, max_log_bytes: usize, ) -> Result { let output = runner.run(&plan.run)?; let logs = capture_command_logs( &plan.virtual_thread, &output.stdout, &output.stderr, max_log_bytes, ); let staged_artifact = if output.status_code == Some(0) { if let Some(path) = plan.stage_stdout_as.take() { Some(overlay.write( path, Digest::sha256(&output.stdout), output.stdout.len() as u64, )) } else { None } } else { None }; if output.status_code == Some(0) { plan.lifecycle.complete(); } Ok(LinuxCommandTaskOutput { virtual_thread: plan.virtual_thread, status_code: output.status_code, stdout: logs.stdout, stderr: logs.stderr, stdout_truncated: logs.stdout_truncated, stderr_truncated: logs.stderr_truncated, log_backpressured: logs.backpressured, staged_artifact, lifecycle: plan.lifecycle, }) } pub fn execute_local_checkout_task( &self, request: LocalCheckoutTaskRequest<'_>, runner: &mut impl ProcessRunner, overlay: &mut VfsOverlay, ) -> Result { let env = request .invocation .env .as_ref() .ok_or(BackendError::MissingEnvironment)?; self.execute_environment_materialization(env, runner)?; let plan = self.plan_local_checkout_run( request.process, request.virtual_thread, request.invocation, request.checkout, request.output_root, request.stage_stdout_as, )?; self.execute_run_plan(plan, runner, overlay) } fn image_tag(&self, env: &clusterflux_core::EnvironmentResource) -> String { let name = env .name .chars() .map(|ch| { if ch.is_ascii_alphanumeric() || ch == '-' || ch == '_' { ch } else { '-' } }) .collect::(); let digest = env .digest .as_str() .chars() .map(|ch| { if ch.is_ascii_alphanumeric() || ch == '-' || ch == '_' || ch == '.' { ch } else { '-' } }) .take(24) .collect::(); format!("clusterflux-env/{name}:{digest}") } } impl CommandBackend for LinuxRootlessPodmanBackend { fn kind(&self) -> CommandBackendKind { CommandBackendKind::LinuxRootlessPodman } fn plan(&self, invocation: &CommandInvocation) -> Result { let env = invocation .env .as_ref() .ok_or(BackendError::MissingEnvironment)?; match env.kind { EnvironmentKind::Containerfile | EnvironmentKind::Dockerfile => Ok(CommandPlan { guest_runtime: GuestRuntimeKind::Wasmtime, backend: CommandBackendKind::LinuxRootlessPodman, required_capability: Capability::RootlessPodman, user_attached_development_execution: false, }), EnvironmentKind::NixFlake => Err(BackendError::UnsupportedEnvironment), } } } #[cfg(test)] mod tests { use std::collections::VecDeque; use std::path::PathBuf; use clusterflux_core::{ DebugRuntimeState, Digest, EnvironmentRequirements, EnvironmentResource, NodeId, ProjectId, TenantId, }; use super::*; #[derive(Default)] struct RecordingRunner { commands: Vec, outputs: VecDeque, } impl RecordingRunner { fn with_outputs(outputs: impl IntoIterator) -> Self { Self { commands: Vec::new(), outputs: outputs.into_iter().collect(), } } } impl ProcessRunner for RecordingRunner { fn run(&mut self, command: &PodmanCommand) -> Result { self.commands.push(command.clone()); self.outputs.pop_front().ok_or_else(|| { BackendError::Command("recording runner has no output queued".to_owned()) }) } } fn success_output(stdout: impl Into>) -> ProcessOutput { ProcessOutput { status_code: Some(0), stdout: stdout.into(), stderr: Vec::new(), } } fn container_env() -> EnvironmentResource { EnvironmentResource { name: "linux".to_owned(), kind: EnvironmentKind::Containerfile, recipe_path: PathBuf::from("envs/linux/Containerfile"), context_path: PathBuf::from("envs/linux"), digest: Digest::sha256("recipe"), requirements: EnvironmentRequirements::linux_container(), } } #[test] fn linux_backend_plans_rootless_podman_under_wasmtime_virtual_task() { let invocation = CommandInvocation { program: "cargo".to_owned(), args: vec!["build".to_owned()], working_directory: "/workspace".to_owned(), environment_variables: Default::default(), timeout_ms: 60_000, network: clusterflux_core::CommandNetworkPolicy::Disabled, env: Some(container_env()), }; let plan = LinuxRootlessPodmanBackend.plan(&invocation).unwrap(); assert_eq!(plan.guest_runtime, GuestRuntimeKind::Wasmtime); assert_eq!(plan.backend, CommandBackendKind::LinuxRootlessPodman); assert_eq!(plan.required_capability, Capability::RootlessPodman); } #[test] fn linux_backend_materializes_containerfile_with_rootless_podman_without_vendored_image() { let env = container_env(); let materialization = LinuxRootlessPodmanBackend .materialize_environment(&env) .unwrap(); assert_eq!(materialization.environment, "linux"); assert!(materialization .image_tag .starts_with("clusterflux-env/linux:")); assert_eq!(materialization.image_tag.matches(':').count(), 1); assert_eq!(materialization.build.program, "podman"); assert!(materialization.rootless_user_podman); assert!(!materialization.embeds_full_image_in_bundle); assert!(materialization.build.args.contains(&"build".to_owned())); assert!(materialization .build .args .contains(&"--pull=missing".to_owned())); assert!(materialization .build .args .contains(&"envs/linux/Containerfile".to_owned())); assert!(materialization .build .args .contains(&"envs/linux".to_owned())); } #[test] fn linux_run_plan_keeps_local_checkout_local_and_avoids_coordinator_file_reads() { let invocation = CommandInvocation { program: "cargo".to_owned(), args: vec!["build".to_owned(), "--release".to_owned()], working_directory: "/workspace/crate".to_owned(), environment_variables: std::collections::BTreeMap::from([( "BUILD_MODE".to_owned(), "release".to_owned(), )]), timeout_ms: 60_000, network: clusterflux_core::CommandNetworkPolicy::Disabled, env: Some(container_env()), }; let plan = LinuxRootlessPodmanBackend .plan_local_checkout_run( ProcessId::from("vp"), TaskInstanceId::from("compile-linux"), &invocation, LocalSourceCheckout { host_path: PathBuf::from("/work/example"), snapshot: Digest::sha256("checkout"), }, PathBuf::from("/work/output"), Some(VfsPath::new("/vfs/artifacts/app").unwrap()), ) .unwrap(); assert_eq!(plan.run.program, "podman"); assert!(plan.run.args.contains(&"run".to_owned())); let name_index = plan .run .args .iter() .position(|argument| argument == "--name") .expect("Podman task has a stable container identity"); assert!(plan.run.args[name_index + 1].starts_with("clusterflux-")); assert!(plan.run.args.contains(&"--network".to_owned())); assert!(plan.run.args.contains(&"none".to_owned())); assert!(plan.run.args.contains(&"--cpus".to_owned())); assert!(plan.run.args.contains(&"--memory".to_owned())); assert!(plan.run.args.contains(&"--pids-limit".to_owned())); assert!(plan.run.args.contains(&"no-new-privileges".to_owned())); assert!(plan.run.args.contains(&"all".to_owned())); assert!(plan.run.args.contains(&"/workspace/crate".to_owned())); assert!(plan.run.args.contains(&"BUILD_MODE=release".to_owned())); assert!(plan .run .args .contains(&"/work/example:/workspace:ro,Z".to_owned())); assert!(plan.run.args.contains(&"cargo".to_owned())); assert!(plan.run.args.contains(&"--release".to_owned())); assert!(!plan.uses_full_repo_tarball); assert!(!plan.coordinator_routed_file_reads); assert!(!plan.source_access.uses_full_repo_tarball()); assert!(!plan.source_access.coordinator_routed_file_reads()); assert_eq!( plan.stage_stdout_as, Some(VfsPath::new("/vfs/artifacts/app").unwrap()) ); assert_eq!(plan.lifecycle.process, ProcessId::from("vp")); assert_eq!( plan.lifecycle.virtual_thread, TaskInstanceId::from("compile-linux") ); assert_eq!(plan.lifecycle.state, LinuxTaskState::Running); match &plan.source_access { SourceAccessMode::LocalCheckoutBindMount { host_path, container_path, read_only, .. } => { assert_eq!(host_path, &PathBuf::from("/work/example")); assert_eq!(container_path, "/workspace"); assert!(*read_only); } SourceAccessMode::NodePreparedSnapshot { .. } => { panic!("local Linux build should use the node-local checkout") } } } #[test] fn linux_backend_executes_podman_build_then_run_and_stages_artifact() { let invocation = CommandInvocation { program: "cargo".to_owned(), args: vec!["build".to_owned()], working_directory: "/workspace".to_owned(), environment_variables: Default::default(), timeout_ms: 60_000, network: clusterflux_core::CommandNetworkPolicy::Disabled, env: Some(container_env()), }; let mut runner = RecordingRunner::with_outputs([success_output([]), success_output(b"artifact-bytes")]); let mut overlay = VfsOverlay::new(TaskInstanceId::from("compile-linux"), NodeId::from("node")); let output = LinuxRootlessPodmanBackend .execute_local_checkout_task( LocalCheckoutTaskRequest { process: ProcessId::from("vp"), virtual_thread: TaskInstanceId::from("compile-linux"), invocation: &invocation, checkout: LocalSourceCheckout { host_path: PathBuf::from("/work/demo"), snapshot: Digest::sha256("checkout"), }, output_root: PathBuf::from("/work/output"), stage_stdout_as: Some(VfsPath::new("/vfs/artifacts/app.tar.zst").unwrap()), }, &mut runner, &mut overlay, ) .unwrap(); let manifest = overlay.flush(); assert_eq!(runner.commands.len(), 2); assert_eq!(runner.commands[0].program, "podman"); assert!(runner.commands[0].args.contains(&"build".to_owned())); assert_eq!(runner.commands[1].program, "podman"); assert!(runner.commands[1].args.contains(&"run".to_owned())); assert!(runner.commands[1] .args .contains(&"/work/demo:/workspace:ro,Z".to_owned())); assert_eq!(output.status_code, Some(0)); assert_eq!(output.stdout, "artifact-bytes"); assert_eq!(output.lifecycle.state, LinuxTaskState::Completed); assert!(output.staged_artifact.is_some()); assert!(manifest .objects .contains_key(&VfsPath::new("/vfs/artifacts/app.tar.zst").unwrap())); assert!(!manifest.large_bytes_uploaded); } #[test] fn linux_backend_caps_logs_without_truncating_staged_artifact_bytes() { let invocation = CommandInvocation { program: "cargo".to_owned(), args: vec!["build".to_owned()], working_directory: "/workspace".to_owned(), environment_variables: Default::default(), timeout_ms: 60_000, network: clusterflux_core::CommandNetworkPolicy::Disabled, env: Some(container_env()), }; let mut runner = RecordingRunner::with_outputs([success_output(b"abcdef"), success_output(b"unused")]); let mut overlay = VfsOverlay::new(TaskInstanceId::from("compile-linux"), NodeId::from("node")); let plan = LinuxRootlessPodmanBackend .plan_local_checkout_run( ProcessId::from("vp"), TaskInstanceId::from("compile-linux"), &invocation, LocalSourceCheckout { host_path: PathBuf::from("/work/demo"), snapshot: Digest::sha256("checkout"), }, PathBuf::from("/work/output"), Some(VfsPath::new("/vfs/artifacts/app.txt").unwrap()), ) .unwrap(); let output = LinuxRootlessPodmanBackend .execute_run_plan_with_log_limit(plan, &mut runner, &mut overlay, 4) .unwrap(); assert_eq!(output.virtual_thread, TaskInstanceId::from("compile-linux")); assert_eq!(output.stdout, "abcd"); assert!(output.stdout_truncated); assert!(output.log_backpressured); assert_eq!(output.staged_artifact.as_ref().unwrap().size, 6); } #[test] fn failed_podman_materialization_returns_clear_error_before_command_run() { let invocation = CommandInvocation { program: "cargo".to_owned(), args: vec!["build".to_owned()], working_directory: "/workspace".to_owned(), environment_variables: Default::default(), timeout_ms: 60_000, network: clusterflux_core::CommandNetworkPolicy::Disabled, env: Some(container_env()), }; let mut runner = RecordingRunner::with_outputs([ProcessOutput { status_code: Some(125), stdout: Vec::new(), stderr: b"image build failed".to_vec(), }]); let mut overlay = VfsOverlay::new(TaskInstanceId::from("compile-linux"), NodeId::from("node")); let error = LinuxRootlessPodmanBackend .execute_local_checkout_task( LocalCheckoutTaskRequest { process: ProcessId::from("vp"), virtual_thread: TaskInstanceId::from("compile-linux"), invocation: &invocation, checkout: LocalSourceCheckout { host_path: PathBuf::from("/work/demo"), snapshot: Digest::sha256("checkout"), }, output_root: PathBuf::from("/work/output"), stage_stdout_as: Some(VfsPath::new("/vfs/artifacts/app.tar.zst").unwrap()), }, &mut runner, &mut overlay, ) .unwrap_err(); assert!(error.to_string().contains("podman build")); assert!(error.to_string().contains("image build failed")); assert_eq!(runner.commands.len(), 1); } #[test] fn linux_task_lifecycle_supports_cancel_and_all_stop_freeze_resume() { let mut lifecycle = LinuxTaskLifecycle::new(ProcessId::from("vp"), TaskInstanceId::from("compile-linux")); lifecycle.freeze_for_debug_epoch().unwrap(); assert_eq!(lifecycle.state, LinuxTaskState::Frozen); lifecycle.resume_after_debug_epoch(); assert_eq!(lifecycle.state, LinuxTaskState::Running); lifecycle.cancel(); assert_eq!(lifecycle.state, LinuxTaskState::Cancelled); let mut unsupported = LinuxTaskLifecycle::new( ProcessId::from("vp"), TaskInstanceId::from("native-command"), ); unsupported.freeze_supported = false; let error = unsupported.freeze_for_debug_epoch().unwrap_err(); assert!(matches!(error, BackendError::DebugFreezeUnsupported { .. })); } #[test] fn windows_backend_is_labeled_user_attached_dev_execution() { let invocation = CommandInvocation { program: "cmd".to_owned(), args: vec!["/C".to_owned(), "build.bat".to_owned()], working_directory: "/workspace".to_owned(), environment_variables: Default::default(), timeout_ms: 60_000, network: clusterflux_core::CommandNetworkPolicy::Disabled, env: None, }; let plan = WindowsCommandDevBackend.plan(&invocation).unwrap(); assert!(plan.user_attached_development_execution); assert_eq!(plan.required_capability, Capability::WindowsCommandDev); } #[test] fn hosted_control_plane_native_command_is_denied() { let error = authorize_node_command(true, true).unwrap_err(); assert!(matches!(error, BackendError::Denied(_))); } #[test] fn native_command_is_denied_without_command_capability() { let error = authorize_node_command(false, false).unwrap_err(); assert!(matches!(error, BackendError::Denied(_))); assert!(error .to_string() .contains("lacks native command capability")); } #[test] fn wasmtime_runtime_runs_named_task_export() { let runtime = WasmtimeTaskRuntime::new().unwrap(); let wasm = r#" (module (func (export "task_add_one") (param i32) (result i32) local.get 0 i32.const 1 i32.add)) "#; let result = runtime.run_i32_export(wasm, "task_add_one", 41).unwrap(); assert_eq!(result, 42); let verified = runtime .run_i32_export_verified(wasm, &Digest::sha256(wasm), "task_add_one", 41) .unwrap(); assert_eq!(verified, 42); } #[test] fn wasmtime_runtime_rejects_bundle_digest_mismatch_before_compilation() { let runtime = WasmtimeTaskRuntime::new().unwrap(); let error = runtime .run_i32_export_verified( "not a valid wasm module", &Digest::sha256("different task bundle bytes"), "task_add_one", 41, ) .unwrap_err(); assert!(matches!(error, WasmTaskError::BundleDigestMismatch { .. })); assert!(!error.to_string().contains("failed to parse")); } #[test] fn wasmtime_runtime_freezes_and_resumes_wasm_debug_participant() { let runtime = WasmtimeTaskRuntime::new().unwrap(); let probe = runtime .freeze_resume_i32_export_probe( r#" (module (func (export "task_add_one") (param i32) (result i32) local.get 0 i32.const 1 i32.add)) "#, "task_add_one", 41, ) .unwrap(); assert_eq!(probe.task, TaskInstanceId::from("task_add_one")); assert_eq!(probe.frozen_state, DebugRuntimeState::Frozen); assert_eq!(probe.resumed_state, DebugRuntimeState::Running); assert_eq!(probe.result, 42); assert!(probe .stack_frames .iter() .any(|frame| frame.contains("task_add_one"))); assert!(probe .local_values .iter() .any(|(name, value)| { name == "wasm_local_0" && value.contains("41") })); assert!(probe.wasm_pc.is_some()); } #[cfg(unix)] #[test] fn native_command_output_is_associated_with_virtual_thread_and_staged_to_vfs() { let executor = LocalCommandExecutor { node: clusterflux_core::NodeId::from("node"), hosted_control_plane: false, has_command_capability: true, }; let mut overlay = VfsOverlay::new(TaskInstanceId::from("compile-linux"), NodeId::from("node")); let output = executor .run( VirtualThreadCommand { virtual_thread: TaskInstanceId::from("compile-linux"), invocation: CommandInvocation { program: "sh".to_owned(), args: vec![ "-c".to_owned(), "printf artifact; printf log >&2".to_owned(), ], working_directory: "/workspace".to_owned(), environment_variables: Default::default(), timeout_ms: 60_000, network: clusterflux_core::CommandNetworkPolicy::Disabled, env: None, }, stage_stdout_as: Some(VfsPath::new("/vfs/artifacts/app.txt").unwrap()), }, &mut overlay, ) .unwrap(); let manifest = overlay.flush(); assert_eq!(output.virtual_thread, TaskInstanceId::from("compile-linux")); assert_eq!(output.stdout, "artifact"); assert_eq!(output.stderr, "log"); assert!(output.staged_artifact.is_some()); assert!(!manifest.large_bytes_uploaded); assert!(manifest .objects .contains_key(&VfsPath::new("/vfs/artifacts/app.txt").unwrap())); } #[cfg(unix)] #[test] fn local_command_executor_caps_logs_and_reports_backpressure_by_virtual_thread() { let executor = LocalCommandExecutor { node: clusterflux_core::NodeId::from("node"), hosted_control_plane: false, has_command_capability: true, }; let mut overlay = VfsOverlay::new(TaskInstanceId::from("compile-linux"), NodeId::from("node")); let output = executor .run_with_log_limit( VirtualThreadCommand { virtual_thread: TaskInstanceId::from("compile-linux"), invocation: CommandInvocation { program: "sh".to_owned(), args: vec!["-c".to_owned(), "printf abcdef; printf err >&2".to_owned()], working_directory: "/workspace".to_owned(), environment_variables: Default::default(), timeout_ms: 60_000, network: clusterflux_core::CommandNetworkPolicy::Disabled, env: None, }, stage_stdout_as: None, }, &mut overlay, 4, ) .unwrap(); assert_eq!(output.virtual_thread, TaskInstanceId::from("compile-linux")); assert_eq!(output.stdout, "abcd"); assert_eq!(output.stderr, ""); assert!(output.stdout_truncated); assert!(output.stderr_truncated); assert!(output.log_backpressured); } #[test] fn public_node_crate_does_not_require_hosted_service_types() { let _tenant = TenantId::from("tenant"); let _project = ProjectId::from("project"); let _backend = LinuxRootlessPodmanBackend; } }