feat(wfe-buildkit, wfe-containerd): add container executor crates

Standalone workspace crates for BuildKit image building and containerd
container execution. Config types, YAML schema integration, compiler
dispatch, validation rules, and mock-based unit tests.

Current implementation shells out to buildctl/nerdctl — will be
replaced with proper gRPC clients (buildkit-client, containerd protos)
in a follow-up. Config types, YAML integration, and test infrastructure
are stable and reusable.

wfe-buildkit: 60 tests, 97.9% library coverage
wfe-containerd: 61 tests, 97.8% library coverage
447 total workspace tests.

wfe-containerd/Cargo.toml

@@ -0,0 +1,23 @@
+[package]
+name = "wfe-containerd"
+version.workspace = true
+edition.workspace = true
+license.workspace = true
+repository.workspace = true
+homepage.workspace = true
+description = "containerd container runner executor for WFE"
+
+[dependencies]
+wfe-core = { workspace = true }
+tokio = { workspace = true }
+serde = { workspace = true }
+serde_json = { workspace = true }
+async-trait = { workspace = true }
+tracing = { workspace = true }
+thiserror = { workspace = true }
+
+[dev-dependencies]
+pretty_assertions = { workspace = true }
+tokio = { workspace = true, features = ["test-util"] }
+tempfile = { workspace = true }
+tokio-util = "0.7"

wfe-containerd/README.md

@@ -0,0 +1,70 @@
+# wfe-containerd
+
+Containerd container runner executor for WFE.
+
+## What it does
+
+`wfe-containerd` runs containers via `nerdctl` as workflow steps. It pulls images, manages registry authentication, and executes containers with configurable networking, resource limits, volume mounts, and TLS settings. Output is captured and parsed for `##wfe[output key=value]` directives, following the same convention as the shell executor.
+
+## Quick start
+
+Add a containerd step to your YAML workflow:
+
+```yaml
+workflow:
+  id: container-pipeline
+  version: 1
+  steps:
+    - name: run-tests
+      type: containerd
+      config:
+        image: node:20-alpine
+        run: npm test
+        network: none
+        memory: 512m
+        cpu: "1.0"
+        timeout: 5m
+        env:
+          NODE_ENV: test
+        volumes:
+          - source: /workspace
+            target: /app
+            readonly: true
+```
+
+Enable the feature in `wfe-yaml`:
+
+```toml
+[dependencies]
+wfe-yaml = { version = "1.0.0", features = ["containerd"] }
+```
+
+## Configuration
+
+| Field | Type | Default | Description |
+|---|---|---|---|
+| `image` | `String` | required | Container image to run |
+| `run` | `String` | - | Shell command (uses `sh -c`) |
+| `command` | `Vec<String>` | - | Command array (mutually exclusive with `run`) |
+| `env` | `HashMap` | `{}` | Environment variables |
+| `volumes` | `Vec<VolumeMount>` | `[]` | Volume mounts |
+| `working_dir` | `String` | - | Working directory inside container |
+| `user` | `String` | `65534:65534` | User/group to run as (nobody by default) |
+| `network` | `String` | `none` | Network mode: `none`, `host`, or `bridge` |
+| `memory` | `String` | - | Memory limit (e.g. `512m`, `1g`) |
+| `cpu` | `String` | - | CPU limit (e.g. `1.0`, `0.5`) |
+| `pull` | `String` | `if-not-present` | Pull policy: `always`, `if-not-present`, `never` |
+| `containerd_addr` | `String` | `/run/containerd/containerd.sock` | Containerd socket address |
+| `tls` | `TlsConfig` | - | TLS configuration for containerd connection |
+| `registry_auth` | `HashMap` | `{}` | Registry authentication per registry hostname |
+| `timeout` | `String` | - | Execution timeout (e.g. `30s`, `5m`) |
+
+## Output parsing
+
+The step captures stdout and stderr. Lines matching `##wfe[output key=value]` are extracted as workflow outputs. Raw stdout, stderr, and exit code are also available under `{step_name}.stdout`, `{step_name}.stderr`, and `{step_name}.exit_code`.
+
+## Security defaults
+
+- Runs as nobody (`65534:65534`) by default
+- Network disabled (`none`) by default
+- Containers are always `--rm` (removed after execution)

wfe-containerd/src/config.rs

@@ -0,0 +1,226 @@
+use std::collections::HashMap;
+
+use serde::{Deserialize, Serialize};
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct ContainerdConfig {
+    pub image: String,
+    pub command: Option<Vec<String>>,
+    pub run: Option<String>,
+    #[serde(default)]
+    pub env: HashMap<String, String>,
+    #[serde(default)]
+    pub volumes: Vec<VolumeMountConfig>,
+    pub working_dir: Option<String>,
+    #[serde(default = "default_user")]
+    pub user: String,
+    #[serde(default = "default_network")]
+    pub network: String,
+    pub memory: Option<String>,
+    pub cpu: Option<String>,
+    #[serde(default = "default_pull")]
+    pub pull: String,
+    #[serde(default = "default_containerd_addr")]
+    pub containerd_addr: String,
+    /// CLI binary name: "nerdctl" (default) or "docker".
+    #[serde(default = "default_cli")]
+    pub cli: String,
+    #[serde(default)]
+    pub tls: TlsConfig,
+    #[serde(default)]
+    pub registry_auth: HashMap<String, RegistryAuth>,
+    pub timeout_ms: Option<u64>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct VolumeMountConfig {
+    pub source: String,
+    pub target: String,
+    #[serde(default)]
+    pub readonly: bool,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, Default)]
+pub struct TlsConfig {
+    pub ca: Option<String>,
+    pub cert: Option<String>,
+    pub key: Option<String>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct RegistryAuth {
+    pub username: String,
+    pub password: String,
+}
+
+fn default_user() -> String {
+    "65534:65534".to_string()
+}
+
+fn default_network() -> String {
+    "none".to_string()
+}
+
+fn default_pull() -> String {
+    "if-not-present".to_string()
+}
+
+fn default_containerd_addr() -> String {
+    "/run/containerd/containerd.sock".to_string()
+}
+
+fn default_cli() -> String {
+    "nerdctl".to_string()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use pretty_assertions::assert_eq;
+
+    #[test]
+    fn serde_round_trip_full_config() {
+        let config = ContainerdConfig {
+            image: "alpine:3.18".to_string(),
+            command: Some(vec!["echo".to_string(), "hello".to_string()]),
+            run: None,
+            env: HashMap::from([("FOO".to_string(), "bar".to_string())]),
+            volumes: vec![VolumeMountConfig {
+                source: "/host/path".to_string(),
+                target: "/container/path".to_string(),
+                readonly: true,
+            }],
+            working_dir: Some("/app".to_string()),
+            user: "1000:1000".to_string(),
+            network: "host".to_string(),
+            memory: Some("512m".to_string()),
+            cpu: Some("1.0".to_string()),
+            pull: "always".to_string(),
+            containerd_addr: "/custom/containerd.sock".to_string(),
+            cli: "nerdctl".to_string(),
+            tls: TlsConfig {
+                ca: Some("/ca.pem".to_string()),
+                cert: Some("/cert.pem".to_string()),
+                key: Some("/key.pem".to_string()),
+            },
+            registry_auth: HashMap::from([(
+                "registry.example.com".to_string(),
+                RegistryAuth {
+                    username: "user".to_string(),
+                    password: "pass".to_string(),
+                },
+            )]),
+            timeout_ms: Some(30000),
+        };
+
+        let json = serde_json::to_string(&config).unwrap();
+        let deserialized: ContainerdConfig = serde_json::from_str(&json).unwrap();
+
+        assert_eq!(deserialized.image, config.image);
+        assert_eq!(deserialized.command, config.command);
+        assert_eq!(deserialized.run, config.run);
+        assert_eq!(deserialized.env, config.env);
+        assert_eq!(deserialized.volumes.len(), 1);
+        assert_eq!(deserialized.volumes[0].source, "/host/path");
+        assert_eq!(deserialized.volumes[0].readonly, true);
+        assert_eq!(deserialized.working_dir, Some("/app".to_string()));
+        assert_eq!(deserialized.user, "1000:1000");
+        assert_eq!(deserialized.network, "host");
+        assert_eq!(deserialized.memory, Some("512m".to_string()));
+        assert_eq!(deserialized.cpu, Some("1.0".to_string()));
+        assert_eq!(deserialized.pull, "always");
+        assert_eq!(deserialized.containerd_addr, "/custom/containerd.sock");
+        assert_eq!(deserialized.tls.ca, Some("/ca.pem".to_string()));
+        assert_eq!(deserialized.tls.cert, Some("/cert.pem".to_string()));
+        assert_eq!(deserialized.tls.key, Some("/key.pem".to_string()));
+        assert!(deserialized.registry_auth.contains_key("registry.example.com"));
+        assert_eq!(deserialized.timeout_ms, Some(30000));
+    }
+
+    #[test]
+    fn serde_round_trip_minimal_config() {
+        let json = r#"{"image": "alpine:latest"}"#;
+        let config: ContainerdConfig = serde_json::from_str(json).unwrap();
+
+        assert_eq!(config.image, "alpine:latest");
+        assert_eq!(config.command, None);
+        assert_eq!(config.run, None);
+        assert!(config.env.is_empty());
+        assert!(config.volumes.is_empty());
+        assert_eq!(config.working_dir, None);
+        assert_eq!(config.user, "65534:65534");
+        assert_eq!(config.network, "none");
+        assert_eq!(config.memory, None);
+        assert_eq!(config.cpu, None);
+        assert_eq!(config.pull, "if-not-present");
+        assert_eq!(config.containerd_addr, "/run/containerd/containerd.sock");
+        assert_eq!(config.timeout_ms, None);
+
+        // Round-trip
+        let serialized = serde_json::to_string(&config).unwrap();
+        let deserialized: ContainerdConfig = serde_json::from_str(&serialized).unwrap();
+        assert_eq!(deserialized.image, "alpine:latest");
+        assert_eq!(deserialized.user, "65534:65534");
+    }
+
+    #[test]
+    fn default_values() {
+        let json = r#"{"image": "busybox"}"#;
+        let config: ContainerdConfig = serde_json::from_str(json).unwrap();
+
+        assert_eq!(config.user, "65534:65534");
+        assert_eq!(config.network, "none");
+        assert_eq!(config.pull, "if-not-present");
+        assert_eq!(config.containerd_addr, "/run/containerd/containerd.sock");
+    }
+
+    #[test]
+    fn volume_mount_serde() {
+        let vol = VolumeMountConfig {
+            source: "/data".to_string(),
+            target: "/mnt/data".to_string(),
+            readonly: false,
+        };
+        let json = serde_json::to_string(&vol).unwrap();
+        let deserialized: VolumeMountConfig = serde_json::from_str(&json).unwrap();
+        assert_eq!(deserialized.source, "/data");
+        assert_eq!(deserialized.target, "/mnt/data");
+        assert_eq!(deserialized.readonly, false);
+
+        // With readonly=true
+        let vol_ro = VolumeMountConfig {
+            source: "/src".to_string(),
+            target: "/dest".to_string(),
+            readonly: true,
+        };
+        let json_ro = serde_json::to_string(&vol_ro).unwrap();
+        let deserialized_ro: VolumeMountConfig = serde_json::from_str(&json_ro).unwrap();
+        assert_eq!(deserialized_ro.readonly, true);
+    }
+
+    #[test]
+    fn tls_config_defaults() {
+        let tls = TlsConfig::default();
+        assert_eq!(tls.ca, None);
+        assert_eq!(tls.cert, None);
+        assert_eq!(tls.key, None);
+
+        let json = r#"{}"#;
+        let deserialized: TlsConfig = serde_json::from_str(json).unwrap();
+        assert_eq!(deserialized.ca, None);
+        assert_eq!(deserialized.cert, None);
+        assert_eq!(deserialized.key, None);
+    }
+
+    #[test]
+    fn registry_auth_serde() {
+        let auth = RegistryAuth {
+            username: "admin".to_string(),
+            password: "secret123".to_string(),
+        };
+        let json = serde_json::to_string(&auth).unwrap();
+        let deserialized: RegistryAuth = serde_json::from_str(&json).unwrap();
+        assert_eq!(deserialized.username, "admin");
+        assert_eq!(deserialized.password, "secret123");
+    }
+}

wfe-containerd/src/lib.rs

@@ -0,0 +1,5 @@
+pub mod config;
+pub mod step;
+
+pub use config::{ContainerdConfig, RegistryAuth, TlsConfig, VolumeMountConfig};
+pub use step::ContainerdStep;

wfe-containerd/tests/integration.rs

@@ -0,0 +1,532 @@
+use std::collections::HashMap;
+use std::io::Write;
+use std::os::unix::fs::PermissionsExt;
+
+use tempfile::TempDir;
+use wfe_containerd::config::{ContainerdConfig, RegistryAuth, TlsConfig};
+use wfe_containerd::ContainerdStep;
+use wfe_core::models::{ExecutionPointer, WorkflowInstance, WorkflowStep};
+use wfe_core::traits::step::{StepBody, StepExecutionContext};
+
+fn minimal_config() -> ContainerdConfig {
+    ContainerdConfig {
+        image: "alpine:3.18".to_string(),
+        command: None,
+        run: Some("echo hello".to_string()),
+        env: HashMap::new(),
+        volumes: vec![],
+        working_dir: None,
+        user: "65534:65534".to_string(),
+        network: "none".to_string(),
+        memory: None,
+        cpu: None,
+        pull: "never".to_string(),
+        containerd_addr: "/run/containerd/containerd.sock".to_string(),
+        cli: "nerdctl".to_string(),
+        tls: TlsConfig::default(),
+        registry_auth: HashMap::new(),
+        timeout_ms: None,
+    }
+}
+
+/// Create a fake nerdctl script in a temp dir.
+fn create_fake_nerdctl(dir: &TempDir, script: &str) -> String {
+    let nerdctl_path = dir.path().join("nerdctl");
+    let mut file = std::fs::File::create(&nerdctl_path).unwrap();
+    file.write_all(script.as_bytes()).unwrap();
+    std::fs::set_permissions(&nerdctl_path, std::fs::Permissions::from_mode(0o755)).unwrap();
+    dir.path().to_string_lossy().to_string()
+}
+
+fn make_context<'a>(
+    step: &'a WorkflowStep,
+    workflow: &'a WorkflowInstance,
+    pointer: &'a ExecutionPointer,
+) -> StepExecutionContext<'a> {
+    StepExecutionContext {
+        item: None,
+        execution_pointer: pointer,
+        persistence_data: None,
+        step,
+        workflow,
+        cancellation_token: tokio_util::sync::CancellationToken::new(),
+    }
+}
+
+/// Wrapper for set_var that handles the Rust 2024 unsafe requirement.
+fn set_path(value: &str) {
+    // SAFETY: These tests run sequentially (nextest runs each test in its own process)
+    // so concurrent mutation of environment variables is not a concern.
+    unsafe {
+        std::env::set_var("PATH", value);
+    }
+}
+
+// ── Happy-path: run succeeds with output parsing ───────────────────
+
+#[tokio::test]
+async fn run_success_with_outputs() {
+    let tmp = TempDir::new().unwrap();
+    let wfe_marker = "##wfe[output";
+    let script = format!(
+        "#!/bin/sh\n\
+         while [ $# -gt 0 ]; do\n\
+         case \"$1\" in\n\
+         run) echo 'hello from container'\n\
+         echo '{wfe_marker} result=success]'\n\
+         echo '{wfe_marker} version=1.0.0]'\n\
+         exit 0;;\n\
+         pull) echo 'Pulling...'; exit 0;;\n\
+         login) exit 0;;\n\
+         --*) shift; shift;;\n\
+         *) shift;;\n\
+         esac\n\
+         done\n\
+         exit 1\n"
+    );
+    let bin_dir = create_fake_nerdctl(&tmp, &script);
+
+    let mut config = minimal_config();
+    config.pull = "always".to_string();
+    let mut step = ContainerdStep::new(config);
+
+    let mut named_step = WorkflowStep::new(0, "containerd");
+    named_step.name = Some("my_container".to_string());
+    let workflow = WorkflowInstance::new("test-wf", 1, serde_json::json!({}));
+    let pointer = ExecutionPointer::new(0);
+    let ctx = make_context(&named_step, &workflow, &pointer);
+
+    let original_path = std::env::var("PATH").unwrap_or_default();
+    set_path(&format!("{bin_dir}:{original_path}"));
+
+    let result = step.run(&ctx).await;
+
+    set_path(&original_path);
+
+    let result = result.expect("run should succeed");
+    assert!(result.proceed);
+    let output = result.output_data.unwrap();
+    let obj = output.as_object().unwrap();
+    assert_eq!(obj.get("result").unwrap(), "success");
+    assert_eq!(obj.get("version").unwrap(), "1.0.0");
+    assert!(
+        obj.get("my_container.stdout")
+            .unwrap()
+            .as_str()
+            .unwrap()
+            .contains("hello from container")
+    );
+    assert_eq!(obj.get("my_container.exit_code").unwrap(), 0);
+}
+
+// ── Non-zero exit code ─────────────────────────────────────────────
+
+#[tokio::test]
+async fn run_container_failure() {
+    let tmp = TempDir::new().unwrap();
+    let script = "#!/bin/sh\n\
+         while [ $# -gt 0 ]; do\n\
+         case \"$1\" in\n\
+         run) echo 'something went wrong' >&2; exit 1;;\n\
+         pull) exit 0;;\n\
+         --*) shift; shift;;\n\
+         *) shift;;\n\
+         esac\n\
+         done\n\
+         exit 1\n";
+    let bin_dir = create_fake_nerdctl(&tmp, script);
+
+    let config = minimal_config();
+    let mut step = ContainerdStep::new(config);
+
+    let wf_step = WorkflowStep::new(0, "containerd");
+    let workflow = WorkflowInstance::new("test-wf", 1, serde_json::json!({}));
+    let pointer = ExecutionPointer::new(0);
+    let ctx = make_context(&wf_step, &workflow, &pointer);
+
+    let original_path = std::env::var("PATH").unwrap_or_default();
+    set_path(&format!("{bin_dir}:{original_path}"));
+
+    let result = step.run(&ctx).await;
+
+    set_path(&original_path);
+
+    let err = result.expect_err("should fail with non-zero exit");
+    let msg = format!("{err}");
+    assert!(msg.contains("Container exited with code 1"), "got: {msg}");
+    assert!(msg.contains("something went wrong"), "got: {msg}");
+}
+
+// ── Pull failure ───────────────────────────────────────────────────
+
+#[tokio::test]
+async fn run_pull_failure() {
+    let tmp = TempDir::new().unwrap();
+    let script = "#!/bin/sh\n\
+         while [ $# -gt 0 ]; do\n\
+         case \"$1\" in\n\
+         pull) echo 'image not found' >&2; exit 1;;\n\
+         --*) shift; shift;;\n\
+         *) shift;;\n\
+         esac\n\
+         done\n\
+         exit 1\n";
+    let bin_dir = create_fake_nerdctl(&tmp, script);
+
+    let mut config = minimal_config();
+    config.pull = "always".to_string();
+    let mut step = ContainerdStep::new(config);
+
+    let wf_step = WorkflowStep::new(0, "containerd");
+    let workflow = WorkflowInstance::new("test-wf", 1, serde_json::json!({}));
+    let pointer = ExecutionPointer::new(0);
+    let ctx = make_context(&wf_step, &workflow, &pointer);
+
+    let original_path = std::env::var("PATH").unwrap_or_default();
+    set_path(&format!("{bin_dir}:{original_path}"));
+
+    let result = step.run(&ctx).await;
+
+    set_path(&original_path);
+
+    let err = result.expect_err("should fail on pull");
+    let msg = format!("{err}");
+    assert!(msg.contains("Image pull failed"), "got: {msg}");
+}
+
+// ── Timeout ────────────────────────────────────────────────────────
+
+#[tokio::test]
+async fn run_timeout() {
+    let tmp = TempDir::new().unwrap();
+    let script = "#!/bin/sh\n\
+         while [ $# -gt 0 ]; do\n\
+         case \"$1\" in\n\
+         run) sleep 30; exit 0;;\n\
+         pull) exit 0;;\n\
+         --*) shift; shift;;\n\
+         *) shift;;\n\
+         esac\n\
+         done\n\
+         exit 1\n";
+    let bin_dir = create_fake_nerdctl(&tmp, script);
+
+    let mut config = minimal_config();
+    config.timeout_ms = Some(100);
+    let mut step = ContainerdStep::new(config);
+
+    let wf_step = WorkflowStep::new(0, "containerd");
+    let workflow = WorkflowInstance::new("test-wf", 1, serde_json::json!({}));
+    let pointer = ExecutionPointer::new(0);
+    let ctx = make_context(&wf_step, &workflow, &pointer);
+
+    let original_path = std::env::var("PATH").unwrap_or_default();
+    set_path(&format!("{bin_dir}:{original_path}"));
+
+    let result = step.run(&ctx).await;
+
+    set_path(&original_path);
+
+    let err = result.expect_err("should timeout");
+    let msg = format!("{err}");
+    assert!(msg.contains("timed out"), "got: {msg}");
+}
+
+// ── Missing nerdctl binary ─────────────────────────────────────────
+
+#[tokio::test]
+async fn run_missing_nerdctl() {
+    let tmp = TempDir::new().unwrap();
+    let bin_dir = tmp.path().to_string_lossy().to_string();
+
+    let config = minimal_config();
+    let mut step = ContainerdStep::new(config);
+
+    let wf_step = WorkflowStep::new(0, "containerd");
+    let workflow = WorkflowInstance::new("test-wf", 1, serde_json::json!({}));
+    let pointer = ExecutionPointer::new(0);
+    let ctx = make_context(&wf_step, &workflow, &pointer);
+
+    let original_path = std::env::var("PATH").unwrap_or_default();
+    set_path(&bin_dir);
+
+    let result = step.run(&ctx).await;
+
+    set_path(&original_path);
+
+    let err = result.expect_err("should fail with missing binary");
+    let msg = format!("{err}");
+    assert!(
+        msg.contains("Failed to spawn nerdctl run") || msg.contains("Failed to pull image")
+            || msg.contains("Failed to spawn docker run"),
+        "got: {msg}"
+    );
+}
+
+// ── pull=never skips pull ──────────────────────────────────────────
+
+#[tokio::test]
+async fn run_skip_pull_when_never() {
+    let tmp = TempDir::new().unwrap();
+    let script = "#!/bin/sh\n\
+         while [ $# -gt 0 ]; do\n\
+         case \"$1\" in\n\
+         run) echo ran; exit 0;;\n\
+         pull) echo 'pull should not be called' >&2; exit 1;;\n\
+         --*) shift; shift;;\n\
+         *) shift;;\n\
+         esac\n\
+         done\n\
+         exit 1\n";
+    let bin_dir = create_fake_nerdctl(&tmp, script);
+
+    let mut config = minimal_config();
+    config.pull = "never".to_string();
+    let mut step = ContainerdStep::new(config);
+
+    let wf_step = WorkflowStep::new(0, "containerd");
+    let workflow = WorkflowInstance::new("test-wf", 1, serde_json::json!({}));
+    let pointer = ExecutionPointer::new(0);
+    let ctx = make_context(&wf_step, &workflow, &pointer);
+
+    let original_path = std::env::var("PATH").unwrap_or_default();
+    set_path(&format!("{bin_dir}:{original_path}"));
+
+    let result = step.run(&ctx).await;
+
+    set_path(&original_path);
+
+    result.expect("should succeed without pulling");
+}
+
+// ── Workflow data is injected as env vars ──────────────────────────
+
+#[tokio::test]
+async fn run_injects_workflow_data_as_env() {
+    let tmp = TempDir::new().unwrap();
+    let script = "#!/bin/sh\n\
+         while [ $# -gt 0 ]; do\n\
+         case \"$1\" in\n\
+         run) shift\n\
+         for arg in \"$@\"; do echo \"ARG:$arg\"; done\n\
+         exit 0;;\n\
+         pull) exit 0;;\n\
+         --*) shift; shift;;\n\
+         *) shift;;\n\
+         esac\n\
+         done\n\
+         exit 1\n";
+    let bin_dir = create_fake_nerdctl(&tmp, script);
+
+    let mut config = minimal_config();
+    config.pull = "never".to_string();
+    config.run = None;
+    config.command = Some(vec!["true".to_string()]);
+    let mut step = ContainerdStep::new(config);
+
+    let mut wf_step = WorkflowStep::new(0, "containerd");
+    wf_step.name = Some("env_test".to_string());
+    let workflow = WorkflowInstance::new(
+        "test-wf",
+        1,
+        serde_json::json!({"my_key": "my_value", "count": 42}),
+    );
+    let pointer = ExecutionPointer::new(0);
+    let ctx = make_context(&wf_step, &workflow, &pointer);
+
+    let original_path = std::env::var("PATH").unwrap_or_default();
+    set_path(&format!("{bin_dir}:{original_path}"));
+
+    let result = step.run(&ctx).await;
+
+    set_path(&original_path);
+
+    let result = result.expect("should succeed");
+    let stdout = result
+        .output_data
+        .unwrap()
+        .as_object()
+        .unwrap()
+        .get("env_test.stdout")
+        .unwrap()
+        .as_str()
+        .unwrap()
+        .to_string();
+
+    // The workflow data should have been injected as uppercase env vars
+    assert!(stdout.contains("MY_KEY=my_value"), "stdout: {stdout}");
+    assert!(stdout.contains("COUNT=42"), "stdout: {stdout}");
+}
+
+// ── Step name defaults to "unknown" when None ──────────────────────
+
+#[tokio::test]
+async fn run_unnamed_step_uses_unknown() {
+    let tmp = TempDir::new().unwrap();
+    let script = "#!/bin/sh\n\
+         while [ $# -gt 0 ]; do\n\
+         case \"$1\" in\n\
+         run) echo output; exit 0;;\n\
+         --*) shift; shift;;\n\
+         *) shift;;\n\
+         esac\n\
+         done\n\
+         exit 1\n";
+    let bin_dir = create_fake_nerdctl(&tmp, script);
+
+    let config = minimal_config();
+    let mut step = ContainerdStep::new(config);
+
+    let wf_step = WorkflowStep::new(0, "containerd");
+    let workflow = WorkflowInstance::new("test-wf", 1, serde_json::json!({}));
+    let pointer = ExecutionPointer::new(0);
+    let ctx = make_context(&wf_step, &workflow, &pointer);
+
+    let original_path = std::env::var("PATH").unwrap_or_default();
+    set_path(&format!("{bin_dir}:{original_path}"));
+
+    let result = step.run(&ctx).await;
+
+    set_path(&original_path);
+
+    let result = result.expect("should succeed");
+    let output = result.output_data.unwrap();
+    let obj = output.as_object().unwrap();
+    assert!(obj.contains_key("unknown.stdout"));
+    assert!(obj.contains_key("unknown.stderr"));
+    assert!(obj.contains_key("unknown.exit_code"));
+}
+
+// ── Registry login failure ─────────────────────────────────────────
+
+#[tokio::test]
+async fn run_login_failure() {
+    let tmp = TempDir::new().unwrap();
+    let script = "#!/bin/sh\n\
+         while [ $# -gt 0 ]; do\n\
+         case \"$1\" in\n\
+         login) echo unauthorized >&2; exit 1;;\n\
+         --*) shift; shift;;\n\
+         *) shift;;\n\
+         esac\n\
+         done\n\
+         exit 0\n";
+    let bin_dir = create_fake_nerdctl(&tmp, script);
+
+    let mut config = minimal_config();
+    config.registry_auth = HashMap::from([(
+        "registry.example.com".to_string(),
+        RegistryAuth {
+            username: "user".to_string(),
+            password: "wrong".to_string(),
+        },
+    )]);
+    let mut step = ContainerdStep::new(config);
+
+    let wf_step = WorkflowStep::new(0, "containerd");
+    let workflow = WorkflowInstance::new("test-wf", 1, serde_json::json!({}));
+    let pointer = ExecutionPointer::new(0);
+    let ctx = make_context(&wf_step, &workflow, &pointer);
+
+    let original_path = std::env::var("PATH").unwrap_or_default();
+    set_path(&format!("{bin_dir}:{original_path}"));
+
+    let result = step.run(&ctx).await;
+
+    set_path(&original_path);
+
+    let err = result.expect_err("should fail on login");
+    let msg = format!("{err}");
+    assert!(msg.contains("login failed"), "got: {msg}");
+}
+
+// ── Successful login with TLS then run ─────────────────────────────
+
+#[tokio::test]
+async fn run_login_success_with_tls() {
+    let tmp = TempDir::new().unwrap();
+    let script = "#!/bin/sh\n\
+         while [ $# -gt 0 ]; do\n\
+         case \"$1\" in\n\
+         run) echo ok; exit 0;;\n\
+         pull) exit 0;;\n\
+         login) exit 0;;\n\
+         --*) shift; shift;;\n\
+         *) shift;;\n\
+         esac\n\
+         done\n\
+         exit 1\n";
+    let bin_dir = create_fake_nerdctl(&tmp, script);
+
+    let mut config = minimal_config();
+    config.pull = "never".to_string();
+    config.tls = TlsConfig {
+        ca: Some("/ca.pem".to_string()),
+        cert: Some("/cert.pem".to_string()),
+        key: Some("/key.pem".to_string()),
+    };
+    config.registry_auth = HashMap::from([(
+        "secure.io".to_string(),
+        RegistryAuth {
+            username: "admin".to_string(),
+            password: "secret".to_string(),
+        },
+    )]);
+    let mut step = ContainerdStep::new(config);
+
+    let wf_step = WorkflowStep::new(0, "containerd");
+    let workflow = WorkflowInstance::new("test-wf", 1, serde_json::json!({}));
+    let pointer = ExecutionPointer::new(0);
+    let ctx = make_context(&wf_step, &workflow, &pointer);
+
+    let original_path = std::env::var("PATH").unwrap_or_default();
+    set_path(&format!("{bin_dir}:{original_path}"));
+
+    let result = step.run(&ctx).await;
+
+    set_path(&original_path);
+
+    result.expect("should succeed with login + TLS");
+}
+
+// ── pull=if-not-present triggers pull ──────────────────────────────
+
+#[tokio::test]
+async fn run_pull_if_not_present() {
+    let tmp = TempDir::new().unwrap();
+    // Track that pull was called via a marker file
+    let marker = tmp.path().join("pull_called");
+    let marker_str = marker.to_string_lossy();
+    let script = format!(
+        "#!/bin/sh\n\
+         while [ $# -gt 0 ]; do\n\
+         case \"$1\" in\n\
+         run) echo ran; exit 0;;\n\
+         pull) touch {marker_str}; exit 0;;\n\
+         --*) shift; shift;;\n\
+         *) shift;;\n\
+         esac\n\
+         done\n\
+         exit 1\n"
+    );
+    let bin_dir = create_fake_nerdctl(&tmp, &script);
+
+    let mut config = minimal_config();
+    config.pull = "if-not-present".to_string();
+    let mut step = ContainerdStep::new(config);
+
+    let wf_step = WorkflowStep::new(0, "containerd");
+    let workflow = WorkflowInstance::new("test-wf", 1, serde_json::json!({}));
+    let pointer = ExecutionPointer::new(0);
+    let ctx = make_context(&wf_step, &workflow, &pointer);
+
+    let original_path = std::env::var("PATH").unwrap_or_default();
+    set_path(&format!("{bin_dir}:{original_path}"));
+
+    let result = step.run(&ctx).await;
+
+    set_path(&original_path);
+
+    result.expect("should succeed");
+    assert!(marker.exists(), "pull should have been called for if-not-present");
+}