adk_code/

container.rs

//! Container code executor — persistent Docker-backed execution environment.
//!
//! [`DockerExecutor`] manages a persistent Docker container that survives across
//! multiple [`execute`](CodeExecutor::execute) calls. Code is written to files
//! inside the container's working directory and executed via `docker exec`,
//! matching the lifecycle model of AutoGen's `DockerCommandLineCodeExecutor`.
//!
//! # Lifecycle
//!
//! ```text
//! DockerExecutor::new(config)
//!     │
//!     ▼
//!   start()  ──► creates & starts container, runs setup commands
//!     │
//!     ▼
//!   execute() ──► writes code to file, exec's inside running container
//!   execute() ──► reuses same container, accumulates workspace state
//!   execute() ──► ...
//!     │
//!     ▼
//!   stop()   ──► stops & removes container
//! ```
//!
//! # Isolation Model
//!
//! The container's network mode, bind mounts, and environment are fixed by
//! [`DockerConfig`] when it is created. Each request's [`SandboxPolicy`] is
//! checked against them with [`DockerConfig::validate_sandbox_policy`], and a
//! request stricter than the container is rejected before any code runs.
//!
//! | Capability | Enforced | Mechanism |
//! |---|---|---|
//! | Network policy | Yes | `NetworkPolicy::Disabled` requires a container created with network mode `none` |
//! | Filesystem policy | Yes | Every bind mount must lie within the paths the policy grants |
//! | Environment policy | Yes | Container variables must be on the allowlist; allowlisted host variables are passed to the exec |
//! | Timeout | Yes | `tokio::time::timeout` on exec |
//! | Structured output | Yes | Last-line JSON extraction from stdout |
//! | Persistent workspace | Yes | Container survives across executions |
//! | Interactive sessions | No | Each execute is a separate exec |
//!
//! # Example
//!
//! ```rust,ignore
//! # async fn example() -> Result<(), adk_code::ExecutionError> {
//! use adk_code::{
//!     CodeExecutor, DockerExecutor, DockerConfig,
//!     ExecutionLanguage, ExecutionPayload, ExecutionRequest, SandboxPolicy,
//! };
//!
//! let executor = DockerExecutor::new(DockerConfig::python())?;
//! executor.start().await?;
//!
//! let result = executor.execute(ExecutionRequest {
//!     language: ExecutionLanguage::Python,
//!     payload: ExecutionPayload::Source {
//!         code: "print('hello from persistent container')".to_string(),
//!     },
//!     argv: vec![],
//!     stdin: None,
//!     input: None,
//!     sandbox: SandboxPolicy::strict_rust(),
//!     identity: None,
//! }).await?;
//!
//! // Container is still running — next execute reuses it
//! executor.stop().await?;
//! # Ok(())
//! # }
//! ```

use std::path::{Component, Path, PathBuf};
use std::process::Stdio;
use std::time::Instant;

use async_trait::async_trait;
use tracing::{debug, info, warn};

use crate::host_process;
use crate::{
    BackendCapabilities, CodeExecutor, EnvironmentPolicy, ExecutionError, ExecutionIsolation,
    ExecutionLanguage, ExecutionPayload, ExecutionRequest, ExecutionResult, ExecutionStatus,
    FilesystemPolicy, NetworkPolicy, SandboxPolicy, validate_request,
};

/// Configuration for the Docker executor.
///
/// # Example
///
/// ```rust
/// use adk_code::DockerConfig;
///
/// let config = DockerConfig::python();
/// assert_eq!(config.image, "python:3.12-slim");
/// assert_eq!(config.work_dir, "/workspace");
/// ```
#[derive(Debug, Clone)]
pub struct DockerConfig {
    /// Container image to use.
    pub image: String,
    /// Working directory inside the container.
    pub work_dir: String,
    /// Setup commands to run after container creation (e.g., `pip install`).
    pub setup_commands: Vec<String>,
    /// Extra environment variables to set in the container.
    pub environment: Vec<String>,
    /// Bind mounts in `host:container[:ro]` format.
    pub bind_mounts: Vec<String>,
    /// Whether to disable network access.
    pub network_disabled: bool,
    /// Container name prefix (a random suffix is appended).
    pub container_name_prefix: String,
    /// Whether to auto-start on first execute if not already running.
    pub auto_start: bool,
    /// Whether to auto-stop and remove the container on drop.
    pub auto_remove: bool,
}

impl DockerConfig {
    /// Python 3.12 preset.
    pub fn python() -> Self {
        Self {
            image: "python:3.12-slim".to_string(),
            work_dir: "/workspace".to_string(),
            setup_commands: vec![],
            environment: vec![],
            bind_mounts: vec![],
            network_disabled: true,
            container_name_prefix: "adk-python".to_string(),
            auto_start: true,
            auto_remove: true,
        }
    }

    /// Node.js 20 preset.
    pub fn node() -> Self {
        Self {
            image: "node:20-slim".to_string(),
            work_dir: "/workspace".to_string(),
            setup_commands: vec![],
            environment: vec![],
            bind_mounts: vec![],
            network_disabled: true,
            container_name_prefix: "adk-node".to_string(),
            auto_start: true,
            auto_remove: true,
        }
    }

    /// Custom image preset.
    pub fn custom(image: impl Into<String>) -> Self {
        Self {
            image: image.into(),
            work_dir: "/workspace".to_string(),
            setup_commands: vec![],
            environment: vec![],
            bind_mounts: vec![],
            network_disabled: true,
            container_name_prefix: "adk-custom".to_string(),
            auto_start: true,
            auto_remove: true,
        }
    }

    /// Add a setup command to run after container creation.
    pub fn setup_command(mut self, cmd: impl Into<String>) -> Self {
        self.setup_commands.push(cmd.into());
        self
    }

    /// Add a pip install command for Python dependencies.
    pub fn pip_install(self, packages: &[&str]) -> Self {
        self.setup_command(format!("pip install --quiet {}", packages.join(" ")))
    }

    /// Add an npm install command for Node.js dependencies.
    pub fn npm_install(self, packages: &[&str]) -> Self {
        self.setup_command(format!("npm install --silent {}", packages.join(" ")))
    }

    /// Enable network access (disabled by default).
    pub fn with_network(mut self) -> Self {
        self.network_disabled = false;
        self
    }

    /// Add a bind mount.
    pub fn bind_mount(mut self, mount: impl Into<String>) -> Self {
        self.bind_mounts.push(mount.into());
        self
    }

    /// Add an environment variable.
    pub fn env(mut self, var: impl Into<String>) -> Self {
        self.environment.push(var.into());
        self
    }

    /// Checks that a container created from this configuration satisfies `policy`.
    ///
    /// A persistent container's network mode, bind mounts, and environment are
    /// fixed when it is created, so a request cannot narrow them afterwards.
    /// `DockerExecutor` calls this before every execution and rejects a request
    /// whose policy is stricter than the container:
    ///
    /// | Policy | Satisfied when |
    /// |---|---|
    /// | `NetworkPolicy::Disabled` | `network_disabled` is `true` |
    /// | `FilesystemPolicy::None` | there are no bind mounts |
    /// | `FilesystemPolicy::WorkspaceReadOnly { root }` | every bind mount is read-only (`:ro`) and its host path lies under `root` |
    /// | `FilesystemPolicy::WorkspaceReadWrite { root }` | every bind mount's host path lies under `root` |
    /// | `FilesystemPolicy::Paths { read_only, read_write }` | every bind mount lies under a `read_write` path, or is read-only and lies under a `read_only` path |
    /// | `EnvironmentPolicy::None` | `environment` is empty |
    /// | `EnvironmentPolicy::AllowList(names)` | every `environment` entry's name is in `names` |
    ///
    /// Bind-mount host paths must be absolute and free of `..` components, so a
    /// named volume never satisfies a filesystem policy. Variables defined by the
    /// image itself (`PATH`, `LANG`) come from the image, not the host, and are
    /// not checked.
    ///
    /// # Errors
    ///
    /// Returns [`ExecutionError::UnsupportedPolicy`] naming the first setting
    /// that exceeds the policy.
    ///
    /// # Example
    ///
    /// ```rust
    /// use adk_code::{DockerConfig, NetworkPolicy, SandboxPolicy};
    ///
    /// let strict = SandboxPolicy::strict_rust();
    /// assert!(DockerConfig::python().validate_sandbox_policy(&strict).is_ok());
    ///
    /// // A networked container cannot honour a request that disables the network.
    /// let networked = DockerConfig::python().with_network();
    /// assert!(networked.validate_sandbox_policy(&strict).is_err());
    ///
    /// let allows_network = SandboxPolicy { network: NetworkPolicy::Enabled, ..strict };
    /// assert!(networked.validate_sandbox_policy(&allows_network).is_ok());
    /// ```
    pub fn validate_sandbox_policy(&self, policy: &SandboxPolicy) -> Result<(), ExecutionError> {
        if matches!(policy.network, NetworkPolicy::Disabled) && !self.network_disabled {
            return Err(ExecutionError::UnsupportedPolicy(
                "the policy disables network access, but the container was created with network \
                 access; drop `with_network()` from the DockerConfig or use \
                 `NetworkPolicy::Enabled`"
                    .to_string(),
            ));
        }

        for mount in &self.bind_mounts {
            // `host:container[:options]`; a Windows drive letter adds a colon to the host part.
            let parts: Vec<&str> = mount.split(':').collect();
            let has_drive_letter = parts.len() > 2
                && parts[0].len() == 1
                && parts[0].chars().all(|c| c.is_ascii_alphabetic());
            let (host, rest) = if has_drive_letter {
                (&mount[..parts[0].len() + 1 + parts[1].len()], &parts[2..])
            } else {
                (parts[0], &parts[1..])
            };
            let read_only = rest
                .get(1)
                .is_some_and(|options| options.split(',').any(|o| o == "ro" || o == "readonly"));

            // `has_root` rather than `is_absolute`: on Windows a Linux-container mount such as
            // `/srv/data` is rooted but not absolute, and containment only needs the path to be
            // anchored and free of `..` for `starts_with` against the policy root to hold.
            let host = Path::new(host);
            let confined = host.has_root()
                && !host.components().any(|component| matches!(component, Component::ParentDir));
            let under = |root: &PathBuf| confined && host.starts_with(root);

            let permitted = match &policy.filesystem {
                FilesystemPolicy::None => false,
                FilesystemPolicy::WorkspaceReadOnly { root } => read_only && under(root),
                FilesystemPolicy::WorkspaceReadWrite { root } => under(root),
                FilesystemPolicy::Paths { read_only: read_only_paths, read_write } => {
                    read_write.iter().any(under) || (read_only && read_only_paths.iter().any(under))
                }
            };
            if !permitted {
                return Err(ExecutionError::UnsupportedPolicy(format!(
                    "bind mount `{mount}` exceeds the filesystem policy {:?}; remove the mount \
                     or grant its host path in the policy",
                    policy.filesystem
                )));
            }
        }

        for entry in &self.environment {
            let name = entry.split_once('=').map_or(entry.as_str(), |(name, _)| name);
            let allowed = match &policy.environment {
                EnvironmentPolicy::None => false,
                EnvironmentPolicy::AllowList(names) => names.iter().any(|allowed| allowed == name),
            };
            if !allowed {
                return Err(ExecutionError::UnsupportedPolicy(format!(
                    "container environment variable `{name}` is not allowed by the environment \
                     policy; remove it from the DockerConfig or add it to \
                     `EnvironmentPolicy::AllowList`"
                )));
            }
        }

        Ok(())
    }
}

impl Default for DockerConfig {
    fn default() -> Self {
        Self::python()
    }
}

// ── Docker SDK implementation (behind `docker` feature) ────────────────

#[cfg(feature = "docker")]
mod docker_impl {
    use super::*;
    use bollard::Docker;
    use bollard::container::{
        Config, CreateContainerOptions, RemoveContainerOptions, StartContainerOptions,
    };
    use bollard::exec::{CreateExecOptions, StartExecResults};
    use futures::StreamExt;
    use rand::Rng;
    use tokio::sync::RwLock;

    /// State of the managed container.
    #[derive(Debug)]
    struct ContainerState {
        /// Docker container ID.
        id: String,
        /// Whether the container is currently running.
        running: bool,
        /// Counter for generating unique filenames.
        file_counter: u64,
    }

    /// Persistent Docker-backed code execution environment.
    ///
    /// Creates a container once via [`start`](CodeExecutor::start), keeps it running,
    /// and executes code via `docker exec` inside the running container. This matches
    /// the lifecycle model of AutoGen's `DockerCommandLineCodeExecutor`.
    ///
    /// # Sandbox Policy
    ///
    /// The container's network mode, bind mounts, and environment come from
    /// [`DockerConfig`] and cannot change per request. [`execute`](CodeExecutor::execute)
    /// rejects with [`ExecutionError::UnsupportedPolicy`] any request whose policy
    /// the container does not satisfy; see [`DockerConfig::validate_sandbox_policy`].
    /// Variables named by an [`EnvironmentPolicy::AllowList`] are passed from the
    /// host to that request's exec.
    ///
    /// # Cleanup
    ///
    /// Call [`cleanup()`](Self::cleanup) explicitly before dropping to ensure
    /// reliable container removal. The [`Drop`] implementation is best-effort
    /// and requires a tokio runtime to be available.
    ///
    /// # Example
    ///
    /// ```rust,no_run
    /// # async fn example() -> Result<(), adk_code::ExecutionError> {
    /// use adk_code::{CodeExecutor, DockerExecutor, DockerConfig};
    ///
    /// let executor = DockerExecutor::new(DockerConfig::python())?;
    /// executor.start().await?;
    /// assert!(executor.is_running().await);
    ///
    /// // Prefer explicit cleanup over relying on Drop
    /// executor.cleanup().await?;
    /// # Ok(())
    /// # }
    /// ```
    pub struct DockerExecutor {
        config: DockerConfig,
        docker: Docker,
        state: RwLock<Option<ContainerState>>,
    }

    impl std::fmt::Debug for DockerExecutor {
        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
            f.debug_struct("DockerExecutor").field("config", &self.config).finish()
        }
    }

    impl DockerExecutor {
        /// Create a new Docker executor with the given configuration.
        ///
        /// This does NOT start the container — call [`start`](CodeExecutor::start)
        /// or set `auto_start: true` (the default) to have it start on first execute.
        /// # Errors
        ///
        /// Returns [`ExecutionError::InternalError`] if the Docker daemon is
        /// unreachable (not installed, not running, or socket permission denied).
        pub fn new(config: DockerConfig) -> std::result::Result<Self, ExecutionError> {
            let docker = Docker::connect_with_local_defaults().map_err(|e| {
                ExecutionError::InternalError(format!(
                    "failed to connect to Docker daemon: {e}. Is Docker installed and running?"
                ))
            })?;
            Ok(Self { config, docker, state: RwLock::new(None) })
        }

        /// Create with a custom Docker connection.
        pub fn with_docker(config: DockerConfig, docker: Docker) -> Self {
            Self { config, docker, state: RwLock::new(None) }
        }

        /// Explicitly stop and remove the Docker container.
        ///
        /// Prefer calling this method before dropping the executor to ensure
        /// reliable cleanup. The [`Drop`] implementation is best-effort and may
        /// not work if no tokio runtime is available.
        pub async fn cleanup(&self) -> Result<(), ExecutionError> {
            let mut state = self.state.write().await;
            if let Some(s) = state.take() {
                info!(container_id = %s.id, "cleaning up container");
                self.docker
                    .remove_container(
                        &s.id,
                        Some(RemoveContainerOptions { force: true, ..Default::default() }),
                    )
                    .await
                    .map_err(|e| {
                        ExecutionError::ExecutionFailed(format!("failed to remove container: {e}"))
                    })?;
            }
            Ok(())
        }

        /// Generate a unique container name.
        fn container_name(&self) -> String {
            let suffix: u32 = rand::rng().random_range(100_000..999_999);
            format!("{}-{suffix}", self.config.container_name_prefix)
        }

        /// Get the file extension for a language.
        fn file_extension(lang: &ExecutionLanguage) -> &'static str {
            match lang {
                ExecutionLanguage::Python => "py",
                ExecutionLanguage::JavaScript => "js",
                ExecutionLanguage::Rust => "rs",
                ExecutionLanguage::Command => "sh",
                ExecutionLanguage::Wasm => "wasm",
            }
        }

        /// Get the execution command for a language and filename.
        fn exec_command(lang: &ExecutionLanguage, filename: &str) -> Vec<String> {
            match lang {
                ExecutionLanguage::Python => {
                    vec!["python3".to_string(), filename.to_string()]
                }
                ExecutionLanguage::JavaScript => {
                    vec!["node".to_string(), filename.to_string()]
                }
                ExecutionLanguage::Command => {
                    vec!["sh".to_string(), filename.to_string()]
                }
                _ => vec![],
            }
        }

        /// Write a file inside the running container.
        async fn write_file(
            &self,
            container_id: &str,
            path: &str,
            content: &str,
        ) -> Result<(), ExecutionError> {
            // Use a heredoc-style approach to write file content safely.
            // Base64 encode to avoid shell escaping issues.
            let encoded = base64_encode(content.as_bytes());
            let cmd = vec![
                "sh".to_string(),
                "-c".to_string(),
                format!("echo '{encoded}' | base64 -d > {path}"),
            ];
            self.exec_in_container(container_id, &cmd, None, None).await?;
            Ok(())
        }

        /// Execute a command inside the running container and capture output.
        ///
        /// `env` entries (`KEY=VALUE`) are added to the container's own environment
        /// for this exec only.
        async fn exec_in_container(
            &self,
            container_id: &str,
            cmd: &[String],
            env: Option<Vec<String>>,
            timeout: Option<std::time::Duration>,
        ) -> Result<(String, String, Option<i64>), ExecutionError> {
            let exec = self
                .docker
                .create_exec(
                    container_id,
                    CreateExecOptions {
                        cmd: Some(cmd.to_vec()),
                        env,
                        attach_stdout: Some(true),
                        attach_stderr: Some(true),
                        working_dir: Some(self.config.work_dir.clone()),
                        ..Default::default()
                    },
                )
                .await
                .map_err(|e| {
                    ExecutionError::ExecutionFailed(format!("failed to create exec: {e}"))
                })?;

            let exec_output = async {
                match self.docker.start_exec(&exec.id, None).await {
                    Ok(StartExecResults::Attached { mut output, .. }) => {
                        let mut stdout = String::new();
                        let mut stderr = String::new();

                        while let Some(chunk) = output.next().await {
                            match chunk {
                                Ok(bollard::container::LogOutput::StdOut { message }) => {
                                    stdout.push_str(&String::from_utf8_lossy(&message));
                                }
                                Ok(bollard::container::LogOutput::StdErr { message }) => {
                                    stderr.push_str(&String::from_utf8_lossy(&message));
                                }
                                Ok(_) => {}
                                Err(e) => {
                                    return Err(ExecutionError::ExecutionFailed(format!(
                                        "exec stream error: {e}"
                                    )));
                                }
                            }
                        }

                        // Get exit code.
                        let inspect = self.docker.inspect_exec(&exec.id).await.map_err(|e| {
                            ExecutionError::ExecutionFailed(format!("failed to inspect exec: {e}"))
                        })?;
                        let exit_code = inspect.exit_code;

                        Ok((stdout, stderr, exit_code))
                    }
                    Ok(StartExecResults::Detached) => Ok((String::new(), String::new(), None)),
                    Err(e) => {
                        Err(ExecutionError::ExecutionFailed(format!("failed to start exec: {e}")))
                    }
                }
            };

            if let Some(dur) = timeout {
                match tokio::time::timeout(dur, exec_output).await {
                    Ok(result) => result,
                    Err(_) => Err(ExecutionError::Timeout(dur.as_millis() as u64)),
                }
            } else {
                exec_output.await
            }
        }
    }

    #[async_trait]
    impl CodeExecutor for DockerExecutor {
        fn name(&self) -> &str {
            "docker"
        }

        fn capabilities(&self) -> BackendCapabilities {
            BackendCapabilities {
                isolation: ExecutionIsolation::ContainerPersistent,
                enforce_network_policy: true,
                enforce_filesystem_policy: true,
                enforce_environment_policy: true,
                enforce_timeout: true,
                supports_structured_output: true,
                supports_process_execution: true,
                supports_persistent_workspace: true,
                supports_interactive_sessions: false,
            }
        }

        fn supports_language(&self, lang: &ExecutionLanguage) -> bool {
            matches!(
                lang,
                ExecutionLanguage::Python
                    | ExecutionLanguage::JavaScript
                    | ExecutionLanguage::Command
            )
        }

        async fn start(&self) -> Result<(), ExecutionError> {
            let mut state = self.state.write().await;
            if state.as_ref().is_some_and(|s| s.running) {
                return Ok(());
            }

            let name = self.container_name();
            info!(image = %self.config.image, container = %name, "creating container");

            // Build container config.
            let mut host_config = bollard::models::HostConfig::default();

            if self.config.network_disabled {
                host_config.network_mode = Some("none".to_string());
            }

            if !self.config.bind_mounts.is_empty() {
                host_config.binds = Some(self.config.bind_mounts.clone());
            }

            let env = if self.config.environment.is_empty() {
                None
            } else {
                Some(self.config.environment.clone())
            };

            let container_config = Config {
                image: Some(self.config.image.clone()),
                working_dir: Some(self.config.work_dir.clone()),
                env,
                host_config: Some(host_config),
                // Keep container alive with a long sleep.
                cmd: Some(vec!["sleep".to_string(), "infinity".to_string()]),
                tty: Some(false),
                ..Default::default()
            };

            let create_opts = CreateContainerOptions { name: name.clone(), ..Default::default() };

            let response =
                self.docker.create_container(Some(create_opts), container_config).await.map_err(
                    |e| ExecutionError::ExecutionFailed(format!("failed to create container: {e}")),
                )?;

            let container_id = response.id;
            debug!(container_id = %container_id, "container created");

            // Start the container.
            self.docker
                .start_container(&container_id, None::<StartContainerOptions<String>>)
                .await
                .map_err(|e| {
                    ExecutionError::ExecutionFailed(format!("failed to start container: {e}"))
                })?;

            info!(container_id = %container_id, "container started");

            // Create workspace directory.
            let mkdir_cmd =
                vec!["mkdir".to_string(), "-p".to_string(), self.config.work_dir.clone()];
            let _ = self.exec_in_container(&container_id, &mkdir_cmd, None, None).await;

            // Run setup commands.
            for setup_cmd in &self.config.setup_commands {
                info!(cmd = %setup_cmd, "running setup command");
                let cmd = vec!["sh".to_string(), "-c".to_string(), setup_cmd.clone()];
                let (_stdout, stderr, exit_code) =
                    self.exec_in_container(&container_id, &cmd, None, None).await?;

                if exit_code != Some(0) {
                    warn!(
                        exit_code = ?exit_code,
                        stderr = %stderr,
                        "setup command failed"
                    );
                    // Clean up on setup failure.
                    let _ = self
                        .docker
                        .remove_container(
                            &container_id,
                            Some(RemoveContainerOptions { force: true, ..Default::default() }),
                        )
                        .await;
                    return Err(ExecutionError::ExecutionFailed(format!(
                        "setup command failed: {setup_cmd}\nstderr: {stderr}"
                    )));
                }
            }

            *state = Some(ContainerState { id: container_id, running: true, file_counter: 0 });

            Ok(())
        }

        async fn stop(&self) -> Result<(), ExecutionError> {
            let mut state = self.state.write().await;
            if let Some(s) = state.take() {
                info!(container_id = %s.id, "stopping container");
                let _ = self
                    .docker
                    .remove_container(
                        &s.id,
                        Some(RemoveContainerOptions { force: true, ..Default::default() }),
                    )
                    .await;
            }
            Ok(())
        }

        async fn is_running(&self) -> bool {
            self.state.read().await.as_ref().is_some_and(|s| s.running)
        }

        async fn execute(
            &self,
            request: ExecutionRequest,
        ) -> Result<ExecutionResult, ExecutionError> {
            let supported = [
                ExecutionLanguage::Python,
                ExecutionLanguage::JavaScript,
                ExecutionLanguage::Command,
            ];
            validate_request(&self.capabilities(), &supported, &request)?;
            self.config.validate_sandbox_policy(&request.sandbox)?;

            let code = match &request.payload {
                ExecutionPayload::Source { code } if code.trim().is_empty() => {
                    return Err(ExecutionError::InvalidRequest("empty source code".to_string()));
                }
                ExecutionPayload::Source { code } => code.clone(),
                ExecutionPayload::GuestModule { .. } => {
                    return Err(ExecutionError::InvalidRequest(
                        "DockerExecutor does not support guest modules".to_string(),
                    ));
                }
            };

            // Auto-start if configured and not running.
            if self.config.auto_start && !self.is_running().await {
                self.start().await?;
            }

            // Get container ID and increment file counter.
            let (container_id, filename) = {
                let mut state = self.state.write().await;
                let s = state.as_mut().ok_or_else(|| {
                    ExecutionError::ExecutionFailed(
                        "container not started — call start() first".to_string(),
                    )
                })?;
                s.file_counter += 1;
                let ext = Self::file_extension(&request.language);
                let filename = format!("{}/code_{}.{ext}", self.config.work_dir, s.file_counter);
                (s.id.clone(), filename)
            };

            let start = Instant::now();

            // Write code to file inside container.
            self.write_file(&container_id, &filename, &code).await?;

            // If there's structured input, write it as a JSON file.
            if let Some(ref input) = request.input {
                let input_json = serde_json::to_string(input).unwrap_or_default();
                let input_path = format!("{}/input.json", self.config.work_dir);
                self.write_file(&container_id, &input_path, &input_json).await?;
            }

            // Build execution command.
            let exec_cmd = Self::exec_command(&request.language, &filename);
            if exec_cmd.is_empty() {
                return Err(ExecutionError::UnsupportedLanguage(format!("{}", request.language)));
            }

            debug!(
                container_id = %container_id,
                language = %request.language,
                filename = %filename,
                "executing code in container"
            );

            // Mirrors `docker run --env NAME`: allowlisted variables carry the host's values.
            let exec_env = host_process::allowlisted_host_env(&request.sandbox.environment)
                .into_iter()
                .map(|(name, value)| format!("{name}={}", value.to_string_lossy()))
                .collect::<Vec<_>>();

            // Execute with timeout.
            let (stdout, stderr, exit_code) = self
                .exec_in_container(
                    &container_id,
                    &exec_cmd,
                    Some(exec_env),
                    Some(request.sandbox.timeout),
                )
                .await?;

            let duration_ms = start.elapsed().as_millis() as u64;

            let (stdout, stdout_truncated) =
                truncate_output(stdout, request.sandbox.max_stdout_bytes);
            let (stderr, stderr_truncated) =
                truncate_output(stderr, request.sandbox.max_stderr_bytes);

            let (structured_output, display_stdout) = extract_structured_output(&stdout);

            let status = match exit_code {
                Some(0) => ExecutionStatus::Success,
                _ => ExecutionStatus::Failed,
            };

            info!(
                exit_code = ?exit_code,
                duration_ms,
                has_structured_output = structured_output.is_some(),
                "container execution completed"
            );

            Ok(ExecutionResult {
                status,
                stdout: display_stdout,
                stderr,
                output: structured_output,
                exit_code: exit_code.map(|c| c as i32),
                stdout_truncated,
                stderr_truncated,
                duration_ms,
                metadata: None,
            })
        }
    }

    impl Drop for DockerExecutor {
        fn drop(&mut self) {
            if self.config.auto_remove {
                // Best-effort cleanup — we can't await in drop, so spawn a task
                // only if a tokio runtime is available.
                if let Some(state) = self.state.get_mut().take() {
                    let docker = self.docker.clone();
                    let container_id = state.id;
                    match tokio::runtime::Handle::try_current() {
                        Ok(handle) => {
                            handle.spawn(async move {
                                let _ = docker
                                    .remove_container(
                                        &container_id,
                                        Some(RemoveContainerOptions {
                                            force: true,
                                            ..Default::default()
                                        }),
                                    )
                                    .await;
                            });
                        }
                        Err(_) => {
                            tracing::warn!(
                                container_id = %container_id,
                                "no tokio runtime available during DockerExecutor drop, \
                                 container may leak. Call cleanup() explicitly before dropping."
                            );
                        }
                    }
                }
            }
        }
    }

    /// Simple base64 encoder (no padding issues with shell).
    fn base64_encode(data: &[u8]) -> String {
        const CHARS: &[u8] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";
        let mut result = String::with_capacity(data.len().div_ceil(3) * 4);
        for chunk in data.chunks(3) {
            let b0 = chunk[0] as u32;
            let b1 = chunk.get(1).copied().unwrap_or(0) as u32;
            let b2 = chunk.get(2).copied().unwrap_or(0) as u32;
            let triple = (b0 << 16) | (b1 << 8) | b2;
            result.push(CHARS[((triple >> 18) & 0x3F) as usize] as char);
            result.push(CHARS[((triple >> 12) & 0x3F) as usize] as char);
            if chunk.len() > 1 {
                result.push(CHARS[((triple >> 6) & 0x3F) as usize] as char);
            } else {
                result.push('=');
            }
            if chunk.len() > 2 {
                result.push(CHARS[(triple & 0x3F) as usize] as char);
            } else {
                result.push('=');
            }
        }
        result
    }
}

#[cfg(feature = "docker")]
pub use docker_impl::DockerExecutor;

// ── CLI-based fallback (always available) ──────────────────────────────

/// Configuration for the CLI-based container command executor.
///
/// This is the fallback executor that shells out to `docker run` for each
/// execution. For production use, prefer `DockerExecutor` (behind the
/// `docker` feature) which uses persistent containers.
///
/// # Example
///
/// ```rust
/// use adk_code::ContainerConfig;
///
/// let config = ContainerConfig::default();
/// assert_eq!(config.runtime, "docker");
/// ```
#[derive(Debug, Clone)]
pub struct ContainerConfig {
    /// Container runtime binary (e.g., `"docker"`, `"podman"`).
    pub runtime: String,
    /// Default container image when not overridden per-request.
    pub default_image: String,
    /// Extra flags passed to the container runtime `run` command.
    pub extra_flags: Vec<String>,
    /// Whether to automatically remove the container after execution.
    pub auto_remove: bool,
}

impl Default for ContainerConfig {
    fn default() -> Self {
        Self {
            runtime: "docker".to_string(),
            default_image: "python:3.12-slim".to_string(),
            extra_flags: vec![],
            auto_remove: true,
        }
    }
}

/// CLI-based container executor that shells out to `docker run` per execution.
///
/// Each [`execute`](CodeExecutor::execute) call spawns a new ephemeral container.
/// This is simpler but less efficient than `DockerExecutor` which reuses a
/// persistent container.
///
/// For production use, prefer `DockerExecutor` (behind the `docker` feature).
///
/// # Example
///
/// ```rust
/// use adk_code::{CodeExecutor, ContainerCommandExecutor, ContainerConfig, ExecutionIsolation};
///
/// let executor = ContainerCommandExecutor::default();
/// assert_eq!(executor.name(), "container-command");
/// assert_eq!(executor.capabilities().isolation, ExecutionIsolation::ContainerEphemeral);
/// ```
#[derive(Debug, Clone)]
pub struct ContainerCommandExecutor {
    config: ContainerConfig,
}

impl ContainerCommandExecutor {
    /// Create a new container command executor with the given configuration.
    pub fn new(config: ContainerConfig) -> Self {
        Self { config }
    }

    /// Build the container `run` command arguments for a given request.
    fn build_run_args(&self, request: &ExecutionRequest) -> Vec<String> {
        let mut args = vec!["run".to_string()];

        if self.config.auto_remove {
            args.push("--rm".to_string());
        }

        args.push("-i".to_string());

        match request.sandbox.network {
            NetworkPolicy::Disabled => {
                args.push("--network=none".to_string());
            }
            NetworkPolicy::Enabled => {}
        }

        match &request.sandbox.filesystem {
            FilesystemPolicy::None => {}
            FilesystemPolicy::WorkspaceReadOnly { root } => {
                args.push("-v".to_string());
                args.push(format!("{}:/workspace:ro", root.display()));
            }
            FilesystemPolicy::WorkspaceReadWrite { root } => {
                args.push("-v".to_string());
                args.push(format!("{}:/workspace:rw", root.display()));
            }
            FilesystemPolicy::Paths { read_only, read_write } => {
                for path in read_only {
                    args.push("-v".to_string());
                    args.push(format!("{}:{}:ro", path.display(), path.display()));
                }
                for path in read_write {
                    args.push("-v".to_string());
                    args.push(format!("{}:{}:rw", path.display(), path.display()));
                }
            }
        }

        if let EnvironmentPolicy::AllowList(vars) = &request.sandbox.environment {
            for var in vars {
                args.push("--env".to_string());
                args.push(var.clone());
            }
        }

        if let Some(ref wd) = request.sandbox.working_directory {
            args.push("-w".to_string());
            args.push(wd.display().to_string());
        }

        args.extend(self.config.extra_flags.clone());
        args.push(self.config.default_image.clone());

        let code = match &request.payload {
            ExecutionPayload::Source { code } => code.clone(),
            ExecutionPayload::GuestModule { .. } => String::new(),
        };

        match request.language {
            ExecutionLanguage::Python => {
                args.push("python3".to_string());
                args.push("-c".to_string());
                args.push(code);
            }
            ExecutionLanguage::JavaScript => {
                args.push("node".to_string());
                args.push("-e".to_string());
                args.push(code);
            }
            ExecutionLanguage::Command => {
                args.push("sh".to_string());
                args.push("-c".to_string());
                args.push(code);
            }
            _ => {}
        }

        args.extend(request.argv.clone());
        args
    }
}

impl Default for ContainerCommandExecutor {
    fn default() -> Self {
        Self::new(ContainerConfig::default())
    }
}

#[async_trait]
impl CodeExecutor for ContainerCommandExecutor {
    fn name(&self) -> &str {
        "container-command"
    }

    fn capabilities(&self) -> BackendCapabilities {
        BackendCapabilities {
            isolation: ExecutionIsolation::ContainerEphemeral,
            enforce_network_policy: true,
            enforce_filesystem_policy: true,
            enforce_environment_policy: true,
            enforce_timeout: true,
            supports_structured_output: true,
            supports_process_execution: true,
            supports_persistent_workspace: false,
            supports_interactive_sessions: false,
        }
    }

    fn supports_language(&self, lang: &ExecutionLanguage) -> bool {
        matches!(
            lang,
            ExecutionLanguage::Python | ExecutionLanguage::JavaScript | ExecutionLanguage::Command
        )
    }

    async fn execute(&self, request: ExecutionRequest) -> Result<ExecutionResult, ExecutionError> {
        let supported =
            [ExecutionLanguage::Python, ExecutionLanguage::JavaScript, ExecutionLanguage::Command];
        validate_request(&self.capabilities(), &supported, &request)?;

        match &request.payload {
            ExecutionPayload::Source { code } if code.trim().is_empty() => {
                return Err(ExecutionError::InvalidRequest("empty source code".to_string()));
            }
            ExecutionPayload::Source { .. } => {}
            ExecutionPayload::GuestModule { .. } => {
                return Err(ExecutionError::InvalidRequest(
                    "ContainerCommandExecutor does not support guest modules".to_string(),
                ));
            }
        }

        let start = Instant::now();
        let run_args = self.build_run_args(&request);

        debug!(
            runtime = %self.config.runtime,
            image = %self.config.default_image,
            language = %request.language,
            "starting container execution"
        );

        // Structured input takes precedence over raw stdin.
        let stdin = match (&request.input, &request.stdin) {
            (Some(input), _) => Some(serde_json::to_vec(input).unwrap_or_default()),
            (None, Some(raw)) => Some(raw.clone()),
            (None, None) => None,
        };

        let mut cmd = tokio::process::Command::new(&self.config.runtime);
        cmd.args(&run_args)
            .stdin(if stdin.is_some() { Stdio::piped() } else { Stdio::null() })
            .stdout(Stdio::piped())
            .stderr(Stdio::piped())
            .kill_on_drop(true);

        let child = cmd.spawn().map_err(|e| {
            ExecutionError::ExecutionFailed(format!(
                "failed to spawn container runtime '{}': {e}",
                self.config.runtime
            ))
        })?;

        let completed = host_process::run_to_completion(
            child,
            stdin,
            request.sandbox.timeout,
            request.sandbox.max_stdout_bytes,
            request.sandbox.max_stderr_bytes,
        )
        .await
        .map_err(|e| {
            ExecutionError::ExecutionFailed(format!("failed to wait for container: {e}"))
        })?;

        let Some(output) = completed else {
            warn!("container execution timed out");
            return Ok(ExecutionResult {
                status: ExecutionStatus::Timeout,
                stdout: String::new(),
                stderr: String::new(),
                output: None,
                exit_code: None,
                stdout_truncated: false,
                stderr_truncated: false,
                duration_ms: start.elapsed().as_millis() as u64,
                metadata: None,
            });
        };

        let duration_ms = start.elapsed().as_millis() as u64;

        let raw_stdout = String::from_utf8_lossy(&output.stdout.bytes).into_owned();
        let raw_stderr = String::from_utf8_lossy(&output.stderr.bytes).into_owned();

        let (stdout, stdout_truncated) =
            truncate_output(raw_stdout, request.sandbox.max_stdout_bytes);
        let (stderr, stderr_truncated) =
            truncate_output(raw_stderr, request.sandbox.max_stderr_bytes);
        let stdout_truncated = stdout_truncated || output.stdout.truncated;
        let stderr_truncated = stderr_truncated || output.stderr.truncated;

        let (structured_output, display_stdout) = extract_structured_output(&stdout);

        let status = if output.status.success() {
            ExecutionStatus::Success
        } else {
            ExecutionStatus::Failed
        };

        info!(
            exit_code = output.status.code(),
            duration_ms,
            has_structured_output = structured_output.is_some(),
            "container execution completed"
        );

        Ok(ExecutionResult {
            status,
            stdout: display_stdout,
            stderr,
            output: structured_output,
            exit_code: output.status.code(),
            stdout_truncated,
            stderr_truncated,
            duration_ms,
            metadata: None,
        })
    }
}

// ── Shared helpers ─────────────────────────────────────────────────────

/// Truncate output to the given byte limit.
fn truncate_output(output: String, max_bytes: usize) -> (String, bool) {
    if output.len() <= max_bytes {
        (output, false)
    } else {
        let truncated = output
            .char_indices()
            .take_while(|(i, _)| *i < max_bytes)
            .map(|(_, c)| c)
            .collect::<String>();
        (truncated, true)
    }
}

/// Extract structured JSON output from the last line of stdout.
fn extract_structured_output(stdout: &str) -> (Option<serde_json::Value>, String) {
    let trimmed = stdout.trim_end();
    if trimmed.is_empty() {
        return (None, String::new());
    }

    if let Some(last_newline_pos) = trimmed.rfind('\n') {
        let last_line = &trimmed[last_newline_pos + 1..];
        let before = &trimmed[..last_newline_pos];

        if let Ok(value) = serde_json::from_str::<serde_json::Value>(last_line) {
            return (Some(value), before.to_string());
        }
    } else if let Ok(value) = serde_json::from_str::<serde_json::Value>(trimmed) {
        return (Some(value), String::new());
    }

    (None, stdout.to_string())
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn capabilities_are_container_ephemeral() {
        let executor = ContainerCommandExecutor::default();
        let caps = executor.capabilities();
        assert_eq!(caps.isolation, ExecutionIsolation::ContainerEphemeral);
        assert!(caps.enforce_network_policy);
        assert!(caps.enforce_filesystem_policy);
        assert!(caps.enforce_environment_policy);
        assert!(caps.enforce_timeout);
        assert!(caps.supports_structured_output);
        assert!(caps.supports_process_execution);
        assert!(!caps.supports_persistent_workspace);
        assert!(!caps.supports_interactive_sessions);
    }

    #[test]
    fn supports_python_js_command() {
        let executor = ContainerCommandExecutor::default();
        assert!(executor.supports_language(&ExecutionLanguage::Python));
        assert!(executor.supports_language(&ExecutionLanguage::JavaScript));
        assert!(executor.supports_language(&ExecutionLanguage::Command));
        assert!(!executor.supports_language(&ExecutionLanguage::Rust));
        assert!(!executor.supports_language(&ExecutionLanguage::Wasm));
    }

    #[test]
    fn default_config() {
        let config = ContainerConfig::default();
        assert_eq!(config.runtime, "docker");
        assert_eq!(config.default_image, "python:3.12-slim");
        assert!(config.extra_flags.is_empty());
        assert!(config.auto_remove);
    }

    #[test]
    fn build_run_args_basic_python() {
        let executor = ContainerCommandExecutor::default();
        let request = ExecutionRequest {
            language: ExecutionLanguage::Python,
            payload: ExecutionPayload::Source { code: "print('hello')".to_string() },
            argv: vec![],
            stdin: None,
            input: None,
            sandbox: crate::SandboxPolicy::strict_rust(),
            identity: None,
        };

        let args = executor.build_run_args(&request);
        assert!(args.contains(&"run".to_string()));
        assert!(args.contains(&"--rm".to_string()));
        assert!(args.contains(&"-i".to_string()));
        assert!(args.contains(&"--network=none".to_string()));
        assert!(args.contains(&"python3".to_string()));
        assert!(args.contains(&"-c".to_string()));
        assert!(args.contains(&"print('hello')".to_string()));
    }

    #[test]
    fn build_run_args_with_network_enabled() {
        let executor = ContainerCommandExecutor::default();
        let mut sandbox = crate::SandboxPolicy::strict_rust();
        sandbox.network = NetworkPolicy::Enabled;

        let request = ExecutionRequest {
            language: ExecutionLanguage::Python,
            payload: ExecutionPayload::Source { code: "print('hello')".to_string() },
            argv: vec![],
            stdin: None,
            input: None,
            sandbox,
            identity: None,
        };

        let args = executor.build_run_args(&request);
        assert!(!args.contains(&"--network=none".to_string()));
    }

    #[test]
    fn docker_config_presets() {
        let py = DockerConfig::python();
        assert_eq!(py.image, "python:3.12-slim");
        assert!(py.network_disabled);

        let node = DockerConfig::node();
        assert_eq!(node.image, "node:20-slim");

        let custom = DockerConfig::custom("ubuntu:24.04");
        assert_eq!(custom.image, "ubuntu:24.04");
    }

    fn policy(
        network: NetworkPolicy,
        filesystem: FilesystemPolicy,
        environment: EnvironmentPolicy,
    ) -> SandboxPolicy {
        SandboxPolicy { network, filesystem, environment, ..SandboxPolicy::strict_rust() }
    }

    fn assert_rejected(config: &DockerConfig, policy: &SandboxPolicy, mentions: &str) {
        match config.validate_sandbox_policy(policy) {
            Err(ExecutionError::UnsupportedPolicy(message)) => {
                assert!(message.contains(mentions), "{message}");
            }
            other => panic!("expected UnsupportedPolicy mentioning `{mentions}`, got {other:?}"),
        }
    }

    #[test]
    fn docker_policy_default_config_satisfies_strict_policy() {
        assert_eq!(
            DockerConfig::python().validate_sandbox_policy(&SandboxPolicy::strict_rust()),
            Ok(())
        );
    }

    #[test]
    fn docker_policy_rejects_disabled_network_on_networked_container() {
        let config = DockerConfig::python().with_network();
        assert_rejected(&config, &SandboxPolicy::strict_rust(), "network");

        let allows_network =
            policy(NetworkPolicy::Enabled, FilesystemPolicy::None, EnvironmentPolicy::None);
        assert_eq!(config.validate_sandbox_policy(&allows_network), Ok(()));
    }

    #[test]
    fn docker_policy_allows_enabled_network_on_isolated_container() {
        let allows_network =
            policy(NetworkPolicy::Enabled, FilesystemPolicy::None, EnvironmentPolicy::None);
        assert_eq!(DockerConfig::python().validate_sandbox_policy(&allows_network), Ok(()));
    }

    #[test]
    fn docker_policy_rejects_any_mount_without_filesystem_access() {
        let config = DockerConfig::python().bind_mount("/srv/data:/data:ro");
        assert_rejected(&config, &SandboxPolicy::strict_rust(), "/srv/data:/data:ro");
    }

    #[test]
    fn docker_policy_read_only_workspace_requires_read_only_mounts_under_root() {
        let read_only = policy(
            NetworkPolicy::Disabled,
            FilesystemPolicy::WorkspaceReadOnly { root: PathBuf::from("/srv") },
            EnvironmentPolicy::None,
        );

        let ro_under_root = DockerConfig::python().bind_mount("/srv/data:/data:ro");
        assert_eq!(ro_under_root.validate_sandbox_policy(&read_only), Ok(()));

        let rw_under_root = DockerConfig::python().bind_mount("/srv/data:/data");
        assert_rejected(&rw_under_root, &read_only, "/srv/data:/data");

        let ro_outside_root = DockerConfig::python().bind_mount("/home/user/.aws:/aws:ro");
        assert_rejected(&ro_outside_root, &read_only, "/home/user/.aws");

        let escapes_root = DockerConfig::python().bind_mount("/srv/../etc:/etc-host:ro");
        assert_rejected(&escapes_root, &read_only, "/srv/../etc");

        let sibling_prefix = DockerConfig::python().bind_mount("/srv-secrets:/s:ro");
        assert_rejected(&sibling_prefix, &read_only, "/srv-secrets");

        let named_volume = DockerConfig::python().bind_mount("cache:/cache:ro");
        assert_rejected(&named_volume, &read_only, "cache:/cache:ro");
    }

    #[test]
    fn docker_policy_read_write_workspace_accepts_any_mode_under_root() {
        let read_write = policy(
            NetworkPolicy::Disabled,
            FilesystemPolicy::WorkspaceReadWrite { root: PathBuf::from("/srv") },
            EnvironmentPolicy::None,
        );
        let config = DockerConfig::python()
            .bind_mount("/srv/data:/data")
            .bind_mount("/srv/models:/models:ro");
        assert_eq!(config.validate_sandbox_policy(&read_write), Ok(()));
    }

    #[test]
    fn docker_policy_explicit_paths_match_mode() {
        let paths = policy(
            NetworkPolicy::Disabled,
            FilesystemPolicy::Paths {
                read_only: vec![PathBuf::from("/srv/models")],
                read_write: vec![PathBuf::from("/srv/out")],
            },
            EnvironmentPolicy::None,
        );

        let config = DockerConfig::python()
            .bind_mount("/srv/models:/models:ro,z")
            .bind_mount("/srv/out:/out:rw");
        assert_eq!(config.validate_sandbox_policy(&paths), Ok(()));

        let writable_read_only_path = DockerConfig::python().bind_mount("/srv/models:/models");
        assert_rejected(&writable_read_only_path, &paths, "/srv/models:/models");
    }

    #[test]
    fn docker_policy_container_environment_must_be_allowlisted() {
        let config = DockerConfig::python().env("API_TOKEN=secret").env("MODE");
        assert_rejected(&config, &SandboxPolicy::strict_rust(), "API_TOKEN");

        let only_mode = policy(
            NetworkPolicy::Disabled,
            FilesystemPolicy::None,
            EnvironmentPolicy::AllowList(vec!["MODE".to_string()]),
        );
        assert_rejected(&config, &only_mode, "API_TOKEN");

        let both = policy(
            NetworkPolicy::Disabled,
            FilesystemPolicy::None,
            EnvironmentPolicy::AllowList(vec!["API_TOKEN".to_string(), "MODE".to_string()]),
        );
        assert_eq!(config.validate_sandbox_policy(&both), Ok(()));
    }

    #[test]
    fn docker_config_builder_methods() {
        let config = DockerConfig::python()
            .pip_install(&["numpy", "pandas"])
            .with_network()
            .env("MY_VAR=hello")
            .bind_mount("/host/data:/data:ro");

        assert!(!config.network_disabled);
        assert_eq!(config.setup_commands.len(), 1);
        assert!(config.setup_commands[0].contains("numpy"));
        assert_eq!(config.environment, vec!["MY_VAR=hello"]);
        assert_eq!(config.bind_mounts, vec!["/host/data:/data:ro"]);
    }
}