clusterflux-public/crates/clusterflux-node/src/lib.rs

use std::path::PathBuf;

use clusterflux_core::{
    Capability, CommandBackendKind, CommandInvocation, CommandPlan, Digest, EnvironmentKind,
    GuestRuntimeKind, ProcessId, TaskInstanceId, VfsObject, VfsOverlay, VfsPath,
};
use serde::{Deserialize, Serialize};
use thiserror::Error;

mod command_runner;
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};

#[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<CommandPlan, BackendError>;
}

#[derive(Clone, Debug, Default)]
pub struct LinuxRootlessPodmanBackend;

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct PodmanCommand {
    pub program: String,
    pub args: Vec<String>,
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
pub struct ProcessOutput {
    pub status_code: Option<i32>,
    pub stdout: Vec<u8>,
    pub stderr: Vec<u8>,
}

pub trait ProcessRunner {
    fn run(&mut self, command: &PodmanCommand) -> Result<ProcessOutput, BackendError>;
}

#[derive(Clone, Debug, Default)]
pub struct StdProcessRunner;

impl ProcessRunner for StdProcessRunner {
    fn run(&mut self, command: &PodmanCommand) -> Result<ProcessOutput, BackendError> {
        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<VfsPath>,
}

#[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<VfsPath>,
    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<i32>,
    pub stdout: String,
    pub stderr: String,
    pub stdout_truncated: bool,
    pub stderr_truncated: bool,
    pub log_backpressured: bool,
    pub staged_artifact: Option<VfsObject>,
    pub lifecycle: LinuxTaskLifecycle,
}

impl LinuxRootlessPodmanBackend {
    pub fn materialize_environment(
        &self,
        env: &clusterflux_core::EnvironmentResource,
    ) -> Result<PodmanEnvironmentMaterialization, BackendError> {
        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<VfsPath>,
    ) -> Result<LinuxCommandRunPlan, BackendError> {
        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 network = match invocation.network {
            clusterflux_core::CommandNetworkPolicy::Disabled => "none",
            clusterflux_core::CommandNetworkPolicy::Enabled => "slirp4netns",
        };
        let mut args = vec![
            "run".to_owned(),
            "--rm".to_owned(),
            "--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<MaterializedEnvironment, BackendError> {
        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<LinuxCommandTaskOutput, BackendError> {
        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<LinuxCommandTaskOutput, BackendError> {
        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<LinuxCommandTaskOutput, BackendError> {
        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::<String>();
        let digest = env
            .digest
            .as_str()
            .chars()
            .map(|ch| {
                if ch.is_ascii_alphanumeric() || ch == '-' || ch == '_' || ch == '.' {
                    ch
                } else {
                    '-'
                }
            })
            .take(24)
            .collect::<String>();
        format!("clusterflux-env/{name}:{digest}")
    }
}

impl CommandBackend for LinuxRootlessPodmanBackend {
    fn kind(&self) -> CommandBackendKind {
        CommandBackendKind::LinuxRootlessPodman
    }

    fn plan(&self, invocation: &CommandInvocation) -> Result<CommandPlan, BackendError> {
        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<PodmanCommand>,
        outputs: VecDeque<ProcessOutput>,
    }

    impl RecordingRunner {
        fn with_outputs(outputs: impl IntoIterator<Item = ProcessOutput>) -> Self {
            Self {
                commands: Vec::new(),
                outputs: outputs.into_iter().collect(),
            }
        }
    }

    impl ProcessRunner for RecordingRunner {
        fn run(&mut self, command: &PodmanCommand) -> Result<ProcessOutput, BackendError> {
            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<Vec<u8>>) -> 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()));
        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_private_types() {
        let _tenant = TenantId::from("tenant");
        let _project = ProjectId::from("project");
        let _backend = LinuxRootlessPodmanBackend;
    }
}