//! Docker-backed sandboxed ToolExecutor.

use async_trait::async_trait;
use serde_json::{json, Value};
use std::path::{Path, PathBuf};
use tokio::sync::Mutex;

/// Ownership label KEY stamped on every CAR sandbox container at create. The
/// VALUE is the creating daemon's PID, so the boot-time reaper
/// ([`reap_orphaned_sandboxes`]) can tell a truly-orphaned container (owner
/// process gone) from one a **still-running** daemon legitimately owns — its
/// own just-created container, or a concurrent observer-mode daemon's live
/// container. Reverse-DNS namespaced so the reaper never matches a foreign
/// container (Parslee-ai/car#479).
pub const SANDBOX_OWNER_LABEL: &str = "ai.parslee.car.sandbox";

/// True if process `pid` is currently alive.
///
/// Unix: `kill(pid, 0)` sends no signal but probes existence — `0` = alive &
/// signalable, `EPERM` = alive but owned by another user (still alive), `ESRCH`
/// = gone.
///
/// Windows (car#516): `OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION)` +
/// `GetExitCodeProcess` — `STILL_ACTIVE` (259) = alive. `OpenProcess` failing
/// with `ERROR_INVALID_PARAMETER` means no such pid → gone; any other error
/// (e.g. access denied) means the process exists but we can't inspect it, so we
/// report alive — the reaper must never remove a live owner's container. (The
/// rare case of a real exit code 259 is treated as alive, an acceptable
/// false-positive for a best-effort reaper.)
fn pid_is_alive(pid: u32) -> bool {
    #[cfg(unix)]
    {
        if pid == 0 {
            return false;
        }
        let rc = unsafe { libc::kill(pid as libc::pid_t, 0) };
        rc == 0 || std::io::Error::last_os_error().raw_os_error() == Some(libc::EPERM)
    }
    #[cfg(target_os = "windows")]
    {
        use windows::Win32::Foundation::{CloseHandle, ERROR_INVALID_PARAMETER, FALSE};
        use windows::Win32::System::Threading::{
            GetExitCodeProcess, OpenProcess, PROCESS_QUERY_LIMITED_INFORMATION,
        };
        const STILL_ACTIVE: u32 = 259;
        if pid == 0 {
            return false;
        }
        unsafe {
            let handle = match OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION, FALSE, pid) {
                Ok(h) => h,
                // No such pid → gone; other errors → assume alive (don't reap).
                Err(e) => return e.code() != ERROR_INVALID_PARAMETER.to_hresult(),
            };
            let mut code: u32 = 0;
            let alive = if GetExitCodeProcess(handle, &mut code).is_ok() {
                code == STILL_ACTIVE
            } else {
                // Couldn't read the exit code; be conservative — treat as alive.
                true
            };
            let _ = CloseHandle(handle);
            alive
        }
    }
    #[cfg(all(not(unix), not(target_os = "windows")))]
    {
        let _ = pid;
        false
    }
}

/// Remove CAR sandbox containers left running by a prior daemon crash/SIGKILL.
///
/// Normal teardown is covered by [`SandboxExecutor`]'s `Drop` (`docker rm -f`),
/// but a hard kill leaves the `sleep infinity` container running forever. This
/// reaper reads each CAR container's owner-PID label value and removes ONLY the
/// ones whose owner is **neither this process nor any live process** — so it
/// never kills the current daemon's own just-created container (boot-race) nor a
/// concurrent daemon's live container (observer/dev mode); it reaps only a dead
/// predecessor's leftovers. Containers from before the owner-PID label existed
/// (value `1`) are treated as orphans and reaped (no CAR daemon runs as PID 1).
///
/// Best-effort: a missing/broken Docker (or nothing to reap) is a no-op, not an
/// error. Returns the number of containers reaped.
pub async fn reap_orphaned_sandboxes() -> usize {
    // Key-only filter (matches regardless of value); we read the owner PID from
    // the value via the ps `--format` template and decide ownership in Rust.
    let list = tokio::process::Command::new("docker")
        .args([
            "ps",
            "-a",
            "--filter",
            &format!("label={SANDBOX_OWNER_LABEL}"),
            "--format",
            &format!("{{{{.ID}}}}|{{{{.Label \"{SANDBOX_OWNER_LABEL}\"}}}}"),
        ])
        .output()
        .await;
    let rows = match list {
        Ok(out) if out.status.success() => String::from_utf8_lossy(&out.stdout).into_owned(),
        _ => return 0,
    };

    let me = std::process::id();
    let to_reap: Vec<String> = rows
        .lines()
        .filter_map(|line| {
            let (id, owner) = line.split_once('|')?;
            let id = id.trim();
            if id.is_empty() {
                return None;
            }
            // Unparseable/empty owner → treat as PID 0 (orphan). Reap unless the
            // owner is us, or a live (>1) process.
            let owner_pid: u32 = owner.trim().parse().unwrap_or(0);
            let owned_by_live_daemon =
                owner_pid == me || (owner_pid > 1 && pid_is_alive(owner_pid));
            (!owned_by_live_daemon).then(|| id.to_string())
        })
        .collect();

    if to_reap.is_empty() {
        return 0;
    }
    let mut args = vec!["rm".to_string(), "-f".to_string()];
    args.extend(to_reap.iter().cloned());
    let _ = tokio::process::Command::new("docker")
        .args(&args)
        .output()
        .await;
    to_reap.len()
}

/// Configuration for a sandbox environment.
#[derive(Debug, Clone)]
pub struct SandboxConfig {
    /// Docker image to use (default: "python:3.11-slim").
    pub image: String,
    /// Working directory to mount into the container.
    pub working_dir: PathBuf,
    /// Additional environment variables.
    pub env: Vec<(String, String)>,
    /// Timeout for individual commands in seconds.
    pub command_timeout_secs: u64,
    /// Whether to keep the container running between commands (faster)
    /// or create a new one per command (more isolated).
    ///
    /// NOT YET WIRED: the executor currently always caches one container
    /// (`ensure_container`). Deliberately not exposed on the
    /// `session.bindSandbox` WS surface until it does something.
    pub persistent: bool,
    /// Memory cap passed to `docker --memory` (e.g. "512m"). `None` = no limit.
    pub memory: Option<String>,
    /// Max process/thread count (`docker --pids-limit`). `None` = no limit.
    pub pids_limit: Option<u64>,
    /// CPU quota (`docker --cpus`, e.g. "1.0"). `None` = no limit.
    pub cpus: Option<String>,
    /// Drop all Linux capabilities and disallow privilege escalation
    /// (`--cap-drop ALL --security-opt no-new-privileges`). Safe for typical
    /// code workloads; on by default.
    pub drop_capabilities: bool,
    /// Read-only root filesystem with a writable tmpfs at `/tmp`. The
    /// `/workspace` mount stays writable. Off by default because some images
    /// need to write outside the workspace (e.g. `pip install`).
    pub read_only_rootfs: bool,
    /// Run container processes as this `UID[:GID]` (`docker --user`). `None` =
    /// the image default (often root). Recommended for untrusted code: set it to
    /// the host user that owns `working_dir` (e.g. `"1000:1000"`) so processes
    /// are non-root AND workspace writes keep correct host ownership. Off by
    /// default because a UID that doesn't own `working_dir` can't write to it.
    pub run_as_user: Option<String>,
    /// Egress policy (D2): the value passed to `docker --network`. `None` =
    /// **`"none"`** — no network, the default-secure posture (D3). Opt IN to
    /// egress with `Some("bridge")` (or any Docker network name) when the
    /// workload genuinely needs it; there is no finer per-host filtering at
    /// this layer (that would require a filtering proxy, out of scope).
    pub network: Option<String>,
}

impl Default for SandboxConfig {
    fn default() -> Self {
        Self {
            image: "python:3.11-slim".into(),
            working_dir: PathBuf::from("."),
            env: Vec::new(),
            command_timeout_secs: 120,
            persistent: true,
            // Hardened-by-default resource caps. These bound a runaway or hostile
            // workload without breaking ordinary code execution.
            memory: Some("512m".into()),
            pids_limit: Some(256),
            cpus: Some("1.0".into()),
            drop_capabilities: true,
            read_only_rootfs: false,
            run_as_user: None,
            network: None, // --network none: no egress unless opted in
        }
    }
}

#[derive(Debug, thiserror::Error)]
pub enum SandboxError {
    #[error("Docker not available: {0}")]
    DockerNotAvailable(String),
    #[error("Container failed: {0}")]
    ContainerFailed(String),
    #[error("Command timeout after {0}s")]
    Timeout(u64),
    #[error("IO error: {0}")]
    Io(#[from] std::io::Error),
}

/// A ToolExecutor that runs shell commands inside a Docker container.
///
/// The working directory is mounted at `/workspace`, and the container runs
/// hardened by default: `--network none`, memory/pids/cpu caps, `--cap-drop ALL`
/// and `--security-opt no-new-privileges` (see [`SandboxConfig`]).
///
/// All ops — shell AND the file ops (`read_file`/`write_file`/`edit_file`/
/// `list_dir`) — execute INSIDE the container (`docker exec`), so the mount
/// namespace is the access boundary: a path cannot reach host files and there is
/// no host path to race (no TOCTOU). Caller paths are mapped to `/workspace/...`
/// by [`container_path`] (absolute paths and `..` rejected) and passed as
/// literal `docker exec` args, so they can't inject shell or options.
///
/// Boundary scope: the host is protected because only `working_dir` is
/// bind-mounted — in-container symlinks resolve within the container and cannot
/// reach the host filesystem. `container_path` confines file ops to `/workspace`
/// *lexically*; it does not chase symlinks created inside the container. That is
/// intentional and not an escalation: the only way to create such a symlink is
/// the in-container shell (`tee` writes regular files), and `exec_shell` already
/// grants full in-container access. The container itself — not `/workspace` — is
/// the trust boundary for untrusted code.
pub struct SandboxExecutor {
    /// Cached `Substrate::name` ("docker:<image>").
    substrate_name: String,
    config: SandboxConfig,
    /// Container ID for persistent mode.
    container_id: Mutex<Option<String>>,
    /// Files that have been read (for read-before-edit enforcement).
    read_files: std::sync::Mutex<std::collections::HashSet<String>>,
}

impl SandboxExecutor {
    /// Create a new sandbox executor. Does NOT start the container yet —
    /// that happens lazily on first command.
    pub fn new(config: SandboxConfig) -> Self {
        let substrate_name = format!("docker:{}", config.image);
        Self {
            config,
            substrate_name,
            container_id: Mutex::new(None),
            read_files: std::sync::Mutex::new(std::collections::HashSet::new()),
        }
    }

    /// Create with just a working directory, using defaults for everything else.
    pub fn for_dir(working_dir: impl Into<PathBuf>) -> Self {
        Self::new(SandboxConfig {
            working_dir: working_dir.into(),
            ..Default::default()
        })
    }

    /// Ensure the container is running, return its ID.
    async fn ensure_container(&self) -> Result<String, String> {
        let mut guard = self.container_id.lock().await;
        if let Some(ref id) = *guard {
            // Check if still running
            let check = tokio::process::Command::new("docker")
                .args(["inspect", "--format", "{{.State.Running}}", id])
                .output()
                .await
                .map_err(|e| format!("docker inspect failed: {}", e))?;
            if String::from_utf8_lossy(&check.stdout).trim() == "true" {
                return Ok(id.clone());
            }
        }

        // Start a new container
        let cwd = std::fs::canonicalize(&self.config.working_dir)
            .unwrap_or_else(|_| self.config.working_dir.clone());

        let container_name = format!(
            "car-sandbox-{}",
            uuid::Uuid::new_v4().to_string().split('-').next().unwrap()
        );

        let mut args = vec![
            "run".to_string(),
            "-d".into(),
            "--name".into(),
            container_name.clone(),
            // Ownership label = the creating daemon's PID, so the boot-time
            // reaper can reclaim this container if the daemon is killed before
            // Drop runs (car#479) yet never touch a container a live daemon
            // (this one, or a concurrent one) still owns.
            "--label".into(),
            format!("{SANDBOX_OWNER_LABEL}={}", std::process::id()),
            "-v".into(),
            format!("{}:/workspace", cwd.display()),
            "-w".into(),
            "/workspace".into(),
            "--network".into(),
            // No network by default for safety; egress is an explicit opt-in
            // via SandboxConfig::network (D2).
            self.config.network.clone().unwrap_or_else(|| "none".into()),
        ];

        // Resource + capability hardening.
        args.extend(hardening_args(&self.config));

        // Add environment variables
        for (k, v) in &self.config.env {
            args.push("-e".into());
            args.push(format!("{}={}", k, v));
        }

        args.push(self.config.image.clone());
        // Keep container alive with a sleep loop
        args.push("sleep".into());
        args.push("infinity".into());

        let output = tokio::process::Command::new("docker")
            .args(&args)
            .output()
            .await
            .map_err(|e| format!("failed to start container: {}", e))?;

        if !output.status.success() {
            let stderr = String::from_utf8_lossy(&output.stderr);
            return Err(format!("docker run failed: {}", stderr));
        }

        let id = String::from_utf8_lossy(&output.stdout).trim().to_string();
        tracing::info!(container_id = %id, image = %self.config.image, "sandbox container started");
        *guard = Some(id.clone());
        Ok(id)
    }

    /// Execute a command inside the container.
    async fn exec_in_container(&self, cmd: &[&str]) -> Result<(String, String, i32), String> {
        let container_id = self.ensure_container().await?;

        let timeout = std::time::Duration::from_secs(self.config.command_timeout_secs);
        let result = tokio::time::timeout(timeout, async {
            tokio::process::Command::new("docker")
                .args(["exec", &container_id])
                .args(cmd)
                .output()
                .await
        })
        .await;

        match result {
            Ok(Ok(output)) => {
                let stdout = String::from_utf8_lossy(&output.stdout).to_string();
                let stderr = String::from_utf8_lossy(&output.stderr).to_string();
                let code = output.status.code().unwrap_or(-1);
                Ok((stdout, stderr, code))
            }
            Ok(Err(e)) => Err(format!("exec failed: {}", e)),
            Err(_) => Err(format!(
                "command timed out after {}s",
                self.config.command_timeout_secs
            )),
        }
    }

    /// Stop and remove the container.
    pub async fn cleanup(&self) {
        let mut guard = self.container_id.lock().await;
        if let Some(ref id) = *guard {
            let _ = tokio::process::Command::new("docker")
                .args(["rm", "-f", id])
                .output()
                .await;
            tracing::info!(container_id = %id, "sandbox container removed");
        }
        *guard = None;
    }

    // --- Tool implementations ---

    async fn exec_shell(&self, params: &Value) -> Result<Value, String> {
        let command = params
            .get("command")
            .and_then(|v| v.as_str())
            .ok_or("missing 'command' parameter")?;

        let (stdout, stderr, exit_code) = self.exec_in_container(&["sh", "-c", command]).await?;

        Ok(json!({
            "stdout": stdout,
            "stderr": stderr,
            "exit_code": exit_code,
        }))
    }

    /// Run a command inside the container, feeding `stdin_data` to its stdin.
    /// Used for writes (`tee`) so content of any size/bytes is delivered without
    /// shell interpolation.
    async fn exec_in_container_stdin(
        &self,
        cmd: &[&str],
        stdin_data: &[u8],
    ) -> Result<(String, String, i32), String> {
        use tokio::io::AsyncWriteExt;
        let container_id = self.ensure_container().await?;
        let timeout = std::time::Duration::from_secs(self.config.command_timeout_secs);

        let run = async {
            let mut child = tokio::process::Command::new("docker")
                .args(["exec", "-i", &container_id])
                .args(cmd)
                .stdin(std::process::Stdio::piped())
                // Discard stdout: `tee` echoes its (possibly large) input to
                // stdout, and we don't read it until after writing all of stdin.
                // Piping it would deadlock once the input exceeds the pipe buffer
                // (~64KB). We only need the file write + exit status + stderr.
                .stdout(std::process::Stdio::null())
                .stderr(std::process::Stdio::piped())
                .spawn()
                .map_err(|e| format!("exec spawn failed: {e}"))?;
            if let Some(mut si) = child.stdin.take() {
                si.write_all(stdin_data)
                    .await
                    .map_err(|e| format!("stdin write: {e}"))?;
                let _ = si.shutdown().await; // close stdin so the writer finishes
            }
            child
                .wait_with_output()
                .await
                .map_err(|e| format!("exec wait: {e}"))
        };

        match tokio::time::timeout(timeout, run).await {
            Ok(Ok(output)) => Ok((
                String::from_utf8_lossy(&output.stdout).to_string(),
                String::from_utf8_lossy(&output.stderr).to_string(),
                output.status.code().unwrap_or(-1),
            )),
            Ok(Err(e)) => Err(e),
            Err(_) => Err(format!(
                "command timed out after {}s",
                self.config.command_timeout_secs
            )),
        }
    }

    async fn exec_read_file(&self, params: &Value) -> Result<Value, String> {
        let path = params
            .get("path")
            .and_then(|v| v.as_str())
            .ok_or("missing 'path' parameter")?;
        let offset = params.get("offset").and_then(|v| v.as_u64()).unwrap_or(0) as usize;
        let limit = params
            .get("limit")
            .and_then(|v| v.as_u64())
            .map(|v| v as usize);

        // Read INSIDE the container so the mount namespace is the boundary — the
        // path can't reach host files, and there is no host path to race
        // (TOCTOU). `--` + literal arg prevents option/shell injection.
        let cpath = container_path(path)?;
        let (content, stderr, code) = self.exec_in_container(&["cat", "--", &cpath]).await?;
        if code != 0 {
            return Err(format!("failed to read file: {}", stderr.trim()));
        }

        let total_lines = content.lines().count();

        let result = if offset > 0 || limit.is_some() {
            let lines: Vec<&str> = content.lines().collect();
            let start = offset.min(lines.len());
            let end = limit
                .map(|l| (start + l).min(lines.len()))
                .unwrap_or(lines.len());
            lines[start..end].join("\n")
        } else {
            content
        };

        // Track as read
        if let Ok(mut set) = self.read_files.lock() {
            set.insert(path.to_string());
        }

        Ok(json!({ "content": result, "total_lines": total_lines }))
    }

    async fn exec_write_file(&self, params: &Value) -> Result<Value, String> {
        let path = params
            .get("path")
            .and_then(|v| v.as_str())
            .ok_or("missing 'path' parameter")?;
        let content = params
            .get("content")
            .and_then(|v| v.as_str())
            .ok_or("missing 'content' parameter")?;

        let cpath = container_path(path)?;
        // Create parent dirs inside the container, then write via `tee` over
        // stdin (binary-safe, no shell interpolation).
        if let Some(parent) = std::path::Path::new(&cpath).parent() {
            let parent = parent.to_string_lossy().to_string();
            let _ = self
                .exec_in_container(&["mkdir", "-p", "--", &parent])
                .await;
        }
        let (_out, stderr, code) = self
            .exec_in_container_stdin(&["tee", "--", &cpath], content.as_bytes())
            .await?;
        if code != 0 {
            return Err(format!("failed to write file: {}", stderr.trim()));
        }

        if let Ok(mut set) = self.read_files.lock() {
            set.insert(path.to_string());
        }

        Ok(json!({ "written": path }))
    }

    async fn exec_edit_file(&self, params: &Value) -> Result<Value, String> {
        let path = params
            .get("path")
            .and_then(|v| v.as_str())
            .ok_or("missing 'path' parameter")?;
        let old_text = params
            .get("old_text")
            .and_then(|v| v.as_str())
            .ok_or("missing 'old_text' parameter")?;
        let new_text = params
            .get("new_text")
            .and_then(|v| v.as_str())
            .ok_or("missing 'new_text' parameter")?;

        let cpath = container_path(path)?;
        let (content, stderr, code) = self.exec_in_container(&["cat", "--", &cpath]).await?;
        if code != 0 {
            return Err(format!("failed to read file: {}", stderr.trim()));
        }

        let count = content.matches(old_text).count();
        if count == 0 {
            // Try whitespace-normalized match
            let normalize = |s: &str| -> String {
                s.lines()
                    .map(|l| l.trim_end())
                    .collect::<Vec<_>>()
                    .join("\n")
            };
            let norm_content = normalize(&content);
            let norm_old = normalize(old_text);
            if norm_content.matches(&norm_old).count() == 1 {
                let norm_pos = norm_content.find(&norm_old).unwrap();
                let start_line = norm_content[..norm_pos].matches('\n').count();
                let old_line_count = old_text.lines().count();
                let actual_lines: Vec<&str> = content.lines().collect();
                let end_line = (start_line + old_line_count).min(actual_lines.len());
                let actual_old = actual_lines[start_line..end_line].join("\n");
                if content.matches(&actual_old).count() == 1 {
                    let new_content = content.replacen(&actual_old, new_text, 1);
                    let (_o, e2, c2) = self
                        .exec_in_container_stdin(&["tee", "--", &cpath], new_content.as_bytes())
                        .await?;
                    if c2 != 0 {
                        return Err(format!("failed to write: {}", e2.trim()));
                    }
                    return Ok(json!({
                        "edited": path,
                        "diff_summary": "whitespace-normalized match",
                    }));
                }
            }

            let first_line = old_text.lines().next().unwrap_or("").trim();
            let mut hint = String::new();
            if !first_line.is_empty() {
                for (i, line) in content.lines().enumerate() {
                    if line.contains(first_line) {
                        let lines: Vec<&str> = content.lines().collect();
                        let start = i.saturating_sub(2);
                        let end = (i + old_text.lines().count() + 2).min(lines.len());
                        hint = format!(
                            "\nActual content at lines {}-{}:\n```\n{}\n```",
                            start + 1,
                            end,
                            lines[start..end].join("\n")
                        );
                        break;
                    }
                }
            }
            return Err(format!("old_text not found in {}.{}", path, hint));
        }
        if count > 1 {
            return Err(format!(
                "old_text found {} times in {} — must be unique",
                count, path
            ));
        }

        let new_content = content.replacen(old_text, new_text, 1);
        let (_o, e2, c2) = self
            .exec_in_container_stdin(&["tee", "--", &cpath], new_content.as_bytes())
            .await?;
        if c2 != 0 {
            return Err(format!("failed to write: {}", e2.trim()));
        }

        let old_lines = old_text.lines().count();
        let new_lines = new_text.lines().count();
        let diff_summary = format!("replaced {} lines with {} lines", old_lines, new_lines);

        Ok(json!({
            "edited": path,
            "diff_summary": diff_summary,
            "new_content": if new_text.len() > 2000 { &new_text[..2000] } else { new_text },
        }))
    }

    async fn exec_list_dir(&self, params: &Value) -> Result<Value, String> {
        let path = params.get("path").and_then(|v| v.as_str()).unwrap_or(".");
        let cpath = container_path(path)?;

        // `ls -1p` inside the container: one entry per line, directories marked
        // with a trailing '/'. Hidden entries are excluded (no -a/-A).
        let (stdout, stderr, code) = self.exec_in_container(&["ls", "-1p", "--", &cpath]).await?;
        if code != 0 {
            return Err(format!("failed to read dir: {}", stderr.trim()));
        }

        let mut entries = Vec::new();
        for line in stdout.lines() {
            if line.is_empty() {
                continue;
            }
            let (name, is_dir) = match line.strip_suffix('/') {
                Some(n) => (n, true),
                None => (line, false),
            };
            entries.push(json!({ "name": name, "is_dir": is_dir }));
        }

        Ok(json!({ "entries": entries }))
    }

    async fn exec_grep_files(&self, params: &Value) -> Result<Value, String> {
        let pattern = params
            .get("pattern")
            .and_then(|v| v.as_str())
            .ok_or("missing 'pattern' parameter")?;
        let path = params.get("path").and_then(|v| v.as_str()).unwrap_or(".");
        // Confine to the workspace (same rule as the file ops) — also makes the
        // search-root always `/workspace/...`, so it can't be a grep option.
        let cpath = container_path(path)?;
        let max_results = params
            .get("max_results")
            .and_then(|v| v.as_u64())
            .unwrap_or(50) as usize;

        // Run grep inside the container. pattern/path/limit are passed as
        // positional args ($1/$2/$3), NOT interpolated into the script, so a
        // crafted pattern/path can't inject shell commands.
        let script = "grep -rn -E \
            --include='*.py' --include='*.rs' --include='*.js' --include='*.ts' \
            --include='*.go' --include='*.java' --include='*.toml' --include='*.json' \
            --include='*.yaml' --include='*.yml' --include='*.html' --include='*.css' \
            --include='*.sh' --include='*.txt' --include='*.cfg' --include='*.ini' \
            -- \"$1\" \"$2\" | head -n \"$3\"";
        let max_str = max_results.to_string();
        let (stdout, _stderr, _code) = self
            .exec_in_container(&["sh", "-c", script, "sh", pattern, &cpath, &max_str])
            .await?;
        let lines: Vec<&str> = stdout.lines().take(max_results).collect();

        Ok(json!({
            "matches": lines,
            "count": lines.len(),
            "truncated": stdout.lines().count() > max_results,
        }))
    }

    async fn exec_find_files(&self, params: &Value) -> Result<Value, String> {
        let pattern = params
            .get("pattern")
            .and_then(|v| v.as_str())
            .ok_or("missing 'pattern' parameter")?;
        let path = params.get("path").and_then(|v| v.as_str()).unwrap_or(".");
        // Confine to the workspace; the search root is then always
        // `/workspace/...`, so it can't be misread as a `find` expression token.
        let cpath = container_path(path)?;

        // path/pattern as positional args ($1/$2) — no shell interpolation.
        let script = "find \"$1\" -name \"$2\" \
            -not -path '*/.*' -not -path '*/node_modules/*' -not -path '*/__pycache__/*' \
            | head -50";
        let (stdout, _stderr, _code) = self
            .exec_in_container(&["sh", "-c", script, "sh", &cpath, pattern])
            .await?;
        let files: Vec<&str> = stdout.lines().collect();

        Ok(json!({ "files": files, "count": files.len() }))
    }
}

#[async_trait]
impl car_engine::ToolExecutor for SandboxExecutor {
    async fn execute(&self, tool: &str, params: &Value) -> Result<Value, String> {
        match tool {
            "shell" => self.exec_shell(params).await,
            "read_file" => self.exec_read_file(params).await,
            "write_file" => self.exec_write_file(params).await,
            "edit_file" => self.exec_edit_file(params).await,
            "list_dir" => self.exec_list_dir(params).await,
            "grep_files" => self.exec_grep_files(params).await,
            "find_files" => self.exec_find_files(params).await,
            "git_status" | "git_diff" | "git_log" | "git_add" | "git_commit" => {
                // Git operations run on host via the mounted volume
                let git_cmd = match tool {
                    "git_status" => "git status --short",
                    "git_diff" => {
                        let staged = params
                            .get("staged")
                            .and_then(|v| v.as_bool())
                            .unwrap_or(false);
                        if staged {
                            "git diff --cached"
                        } else {
                            "git diff"
                        }
                    }
                    "git_log" => "git log --oneline -20",
                    _ => return Err(format!("git tool {} not implemented in sandbox", tool)),
                };
                self.exec_shell(&json!({ "command": git_cmd })).await
            }
            _ => Err(format!("unknown tool in sandbox: {}", tool)),
        }
    }
}

impl Drop for SandboxExecutor {
    fn drop(&mut self) {
        // Best-effort cleanup — can't async in Drop, so use blocking
        if let Ok(guard) = self.container_id.try_lock() {
            if let Some(ref id) = *guard {
                let _ = std::process::Command::new("docker")
                    .args(["rm", "-f", id])
                    .output();
            }
        }
    }
}

/// Map a caller-supplied path to an absolute in-container path under
/// `/workspace`. Rejects absolute paths and any `..` component, so the result is
/// always a plain workspace-relative path. The container mount namespace is the
/// real boundary — this is pure string mapping, with no host filesystem access
/// and therefore no check-then-use (TOCTOU) window.
fn container_path(path: &str) -> Result<String, String> {
    if path.is_empty() {
        return Err("path must not be empty (use \".\" for the workspace root)".to_string());
    }
    // Container paths are POSIX. Parse them platform-independently rather than
    // via `std::path` — on Windows `Path::new("/etc").is_absolute()` is `false`
    // (no drive letter), which would let a leading-slash path slip past the
    // is-absolute guard (car#516 follow-up). A leading `/` or `\`, or a
    // Windows drive-absolute path (`C:\…`), is absolute here.
    if path.starts_with('/') || path.starts_with('\\') || Path::new(path).is_absolute() {
        return Err(format!("path '{path}' must be relative to the workspace"));
    }
    let mut out = String::from("/workspace");
    // Split on both separators so a Windows-style `sub\dir` is handled too.
    for seg in path.split(['/', '\\']) {
        match seg {
            "" | "." => {}
            ".." => return Err(format!("path '{path}' may not contain '..'")),
            normal => {
                out.push('/');
                out.push_str(normal);
            }
        }
    }
    Ok(out)
}

/// Build the resource/capability `docker run` flags for a config. Pure so the
/// hardening profile is unit-testable without invoking docker.
fn hardening_args(config: &SandboxConfig) -> Vec<String> {
    let mut args = Vec::new();
    if let Some(mem) = &config.memory {
        args.push("--memory".into());
        args.push(mem.clone());
        // Pin swap to the memory limit so the cap can't be evaded via swap.
        args.push("--memory-swap".into());
        args.push(mem.clone());
    }
    if let Some(pids) = config.pids_limit {
        args.push("--pids-limit".into());
        args.push(pids.to_string());
    }
    if let Some(cpus) = &config.cpus {
        args.push("--cpus".into());
        args.push(cpus.clone());
    }
    if config.drop_capabilities {
        args.push("--cap-drop".into());
        args.push("ALL".into());
        args.push("--security-opt".into());
        args.push("no-new-privileges".into());
    }
    if config.read_only_rootfs {
        args.push("--read-only".into());
        // Rootfs is read-only, so give the workload a writable scratch area
        // (the /workspace bind mount stays writable independently).
        args.push("--tmpfs".into());
        args.push("/tmp:rw,size=512m".into());
    }
    if let Some(user) = &config.run_as_user {
        args.push("--user".into());
        args.push(user.clone());
    }
    args
}

// ---------------------------------------------------------------------------
// Substrate impl (D1): the sandbox as a selectable EXECUTION ENVIRONMENT,
// not just a tool executor. Binding it via `Runtime::set_substrate` routes
// the commodity built-ins (shell / read_file / write_file / list_dir /
// grep_files composition) INTO the container — the same pattern as
// `McpSubstrate` for `session.bindSubstrate`, so the daemon can offer
// `session.bindSandbox`. All paths are container paths under /workspace
// (host reach is impossible by mount-namespace construction); commands run
// with the container's hardened profile (no network unless opted in,
// resource caps, dropped capabilities).
// ---------------------------------------------------------------------------

#[async_trait::async_trait]
impl car_engine::Substrate for SandboxExecutor {
    fn name(&self) -> &str {
        &self.substrate_name
    }

    async fn run_command(
        &self,
        cmd: &str,
        timeout_s: Option<f64>,
    ) -> Result<car_engine::CommandOutput, String> {
        // Per-call timeout override: the trait's contract beats the
        // config default when the caller supplies one.
        let container_id = self.ensure_container().await?;
        // Trait contract alignment (review Q4): like LocalSubstrate,
        // `Some(t)` with t <= 0 means UNLIMITED, not an instant timeout.
        let secs = match timeout_s {
            Some(t) if t > 0.0 => t,
            Some(_) => f64::INFINITY,
            None => self.config.command_timeout_secs as f64,
        };
        let timeout = if secs.is_finite() {
            std::time::Duration::from_secs_f64(secs)
        } else {
            std::time::Duration::MAX
        };
        // kill_on_drop stops the HOST docker-exec client when the timeout
        // drops the future. Known, owned limitation (review Q2): the
        // in-container process itself keeps running — docker has no
        // exec-kill API short of killing the container — so repeated
        // timeouts can accumulate processes against pids_limit. A caller
        // that needs the workload dead wraps it in `timeout N ...` inside
        // the command (image-dependent), or re-binds with a fresh sandbox.
        let result = tokio::time::timeout(timeout, async {
            tokio::process::Command::new("docker")
                .args(["exec", &container_id, "sh", "-c", cmd])
                .kill_on_drop(true)
                .output()
                .await
        })
        .await;
        match result {
            Ok(Ok(output)) => Ok(car_engine::CommandOutput {
                stdout: String::from_utf8_lossy(&output.stdout).to_string(),
                stderr: String::from_utf8_lossy(&output.stderr).to_string(),
                exit_code: output.status.code().unwrap_or(-1),
            }),
            Ok(Err(e)) => Err(format!("exec failed: {e}")),
            Err(_) => Err(format!("command timed out after {secs}s")),
        }
    }

    async fn read_text(&self, path: &str) -> Result<String, String> {
        let cpath = container_path(path)?;
        let (content, stderr, code) = self.exec_in_container(&["cat", "--", &cpath]).await?;
        if code != 0 {
            return Err(format!("failed to read file: {}", stderr.trim()));
        }
        Ok(content)
    }

    async fn write_text(&self, path: &str, content: &str) -> Result<(), String> {
        self.substrate_write(path, content.as_bytes()).await
    }

    async fn read_bytes(
        &self,
        path: &str,
        offset: Option<u64>,
        len: Option<u64>,
    ) -> Result<Vec<u8>, String> {
        use base64::Engine;
        let cpath = container_path(path)?;
        // Read INSIDE the container and base64 the bytes out — `docker exec`
        // output is a text pipe (from_utf8_lossy would corrupt raw bytes).
        // tail/head do the windowing so `dd bs=1` costs are avoided.
        let window = match (offset, len) {
            (None, None) => format!("base64 < {}", shell_quote(&cpath)),
            (off, l) => {
                let skip = off.unwrap_or(0);
                let head = l.map(|n| format!(" | head -c {n}")).unwrap_or_default();
                // Existence guard (review C2): a pipeline's exit status is
                // the LAST command's, and base64 exits 0 on empty input —
                // so `tail missing | base64` returned Ok(empty), a silent
                // wrong answer. Guard readability up front instead.
                format!(
                    "[ -r {p} ] || exit 66; tail -c +{} {p}{} | base64",
                    skip.saturating_add(1),
                    head,
                    p = shell_quote(&cpath),
                )
            }
        };

        let (b64, stderr, code) = self.exec_in_container(&["sh", "-c", &window]).await?;
        if code != 0 {
            return Err(format!("failed to read file: {}", stderr.trim()));
        }
        let cleaned: String = b64.split_whitespace().collect();
        base64::engine::general_purpose::STANDARD
            .decode(cleaned.as_bytes())
            .map_err(|e| format!("base64 decode: {e}"))
    }

    async fn write_bytes(&self, path: &str, bytes: &[u8]) -> Result<(), String> {
        self.substrate_write(path, bytes).await
    }
}

impl SandboxExecutor {
    /// Shared write path for the Substrate impl: mkdir -p the parent, then
    /// stream bytes over stdin via `tee` (binary-safe, no interpolation) —
    /// the same technique as `exec_write_file`.
    async fn substrate_write(&self, path: &str, bytes: &[u8]) -> Result<(), String> {
        let cpath = container_path(path)?;
        if let Some(parent) = std::path::Path::new(&cpath).parent() {
            let parent = parent.to_string_lossy().to_string();
            // Propagated (review Q6): swallowing this hid "docker daemon
            // not running" behind tee's less useful error. `mkdir -p` on
            // an existing dir exits 0, so propagation is side-effect-free.
            self.exec_in_container(&["mkdir", "-p", "--", &parent])
                .await?;
        }
        let (_, stderr, code) = self
            .exec_in_container_stdin(&["tee", "--", &cpath], bytes)
            .await?;
        if code != 0 {
            return Err(format!("failed to write file: {}", stderr.trim()));
        }
        Ok(())
    }
}

/// Single-quote a string for `sh -c` (the only metacharacter inside single
/// quotes is the single quote itself).
fn shell_quote(s: &str) -> String {
    format!("'{}'", s.replace('\'', "'\\''"))
}

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

    #[test]
    fn owner_label_is_namespaced() {
        // The reaper filters on this exact label; keep it CAR-namespaced so it
        // can never match a foreign container (car#479).
        assert_eq!(SANDBOX_OWNER_LABEL, "ai.parslee.car.sandbox");
    }

    #[tokio::test]
    async fn reap_is_a_noop_without_orphans() {
        // Best-effort contract: with no CAR containers present (and/or no
        // Docker on this host), the reaper reaps nothing and never errors.
        assert_eq!(reap_orphaned_sandboxes().await, 0);
    }

    #[test]
    fn pid_liveness_probe() {
        // This process is alive; a PID that cannot exist is not.
        assert!(pid_is_alive(std::process::id()));
        assert!(!pid_is_alive(0));
        // The reaper's ownership rule: our own PID and any live PID are spared;
        // owner "1" (pre-label-value orphans) and dead/unparseable owners reap.
        let me = std::process::id();
        let owned_by_live = |owner: u32| owner == me || (owner > 1 && pid_is_alive(owner));
        assert!(owned_by_live(me)); // our container — spared
        assert!(!owned_by_live(1)); // legacy "=1" orphan — reaped
        assert!(!owned_by_live(0)); // unparseable owner — reaped
    }

    #[test]
    fn sandbox_config_defaults() {
        let config = SandboxConfig::default();
        assert_eq!(config.image, "python:3.11-slim");
        assert!(config.persistent);
        assert_eq!(config.command_timeout_secs, 120);
        // Hardened by default.
        assert_eq!(config.memory.as_deref(), Some("512m"));
        assert_eq!(config.pids_limit, Some(256));
        assert_eq!(config.cpus.as_deref(), Some("1.0"));
        assert!(config.drop_capabilities);
        assert!(!config.read_only_rootfs);
    }

    #[test]
    fn default_hardening_flags_are_emitted() {
        let args = hardening_args(&SandboxConfig::default());
        let joined = args.join(" ");
        assert!(joined.contains("--memory 512m"));
        assert!(joined.contains("--memory-swap 512m"));
        assert!(joined.contains("--pids-limit 256"));
        assert!(joined.contains("--cpus 1.0"));
        assert!(joined.contains("--cap-drop ALL"));
        assert!(joined.contains("--security-opt no-new-privileges"));
        // Read-only rootfs is opt-in.
        assert!(!joined.contains("--read-only"));
    }

    #[test]
    fn run_as_user_emitted_when_set() {
        let config = SandboxConfig {
            run_as_user: Some("1000:1000".into()),
            ..Default::default()
        };
        let joined = hardening_args(&config).join(" ");
        assert!(joined.contains("--user 1000:1000"));
        // Off by default.
        assert!(!hardening_args(&SandboxConfig::default())
            .join(" ")
            .contains("--user"));
    }

    #[test]
    fn read_only_rootfs_adds_tmpfs_when_enabled() {
        let config = SandboxConfig {
            read_only_rootfs: true,
            ..Default::default()
        };
        let joined = hardening_args(&config).join(" ");
        assert!(joined.contains("--read-only"));
        assert!(joined.contains("--tmpfs /tmp:rw,size=512m"));
    }

    #[test]
    fn limits_can_be_disabled() {
        let config = SandboxConfig {
            memory: None,
            pids_limit: None,
            cpus: None,
            drop_capabilities: false,
            ..Default::default()
        };
        assert!(hardening_args(&config).is_empty());
    }

    #[test]
    fn container_path_maps_relative_paths_under_workspace() {
        assert_eq!(container_path("file.txt").unwrap(), "/workspace/file.txt");
        assert_eq!(
            container_path("sub/dir/file.txt").unwrap(),
            "/workspace/sub/dir/file.txt"
        );
        // `.` and `./` normalize to the workspace root.
        assert_eq!(container_path(".").unwrap(), "/workspace");
        assert_eq!(container_path("./a").unwrap(), "/workspace/a");
    }

    #[test]
    fn container_path_rejects_absolute_and_traversal() {
        assert!(container_path("/etc/passwd")
            .unwrap_err()
            .contains("must be relative"));
        assert!(container_path("../../etc/passwd")
            .unwrap_err()
            .contains("'..'"));
        assert!(container_path("sub/../../etc")
            .unwrap_err()
            .contains("'..'"));
        assert!(container_path("")
            .unwrap_err()
            .contains("must not be empty"));
    }
}

#[cfg(test)]
mod substrate_live_tests {
    use super::*;
    use car_engine::Substrate;

    /// Live Docker round-trip of the D1 Substrate impl. `--ignored`: needs a
    /// running Docker daemon (run manually / on a Docker-equipped host).
    #[tokio::test]
    #[ignore = "requires a running Docker daemon"]
    async fn substrate_roundtrip_in_container() {
        let tmp = tempfile::tempdir().unwrap();
        let ex = SandboxExecutor::new(SandboxConfig {
            image: "alpine:3.20".into(),
            working_dir: tmp.path().to_path_buf(),
            command_timeout_secs: 60,
            ..Default::default()
        });
        assert_eq!(ex.name(), "docker:alpine:3.20");

        // run_command with a per-call timeout.
        let out = ex
            .run_command("echo hello && uname", Some(30.0))
            .await
            .unwrap();
        assert_eq!(out.exit_code, 0);
        assert!(out.stdout.contains("hello"));
        assert!(
            out.stdout.contains("Linux"),
            "runs in the container, not the host"
        );

        // write_text / read_text round-trip (path is container-relative).
        ex.write_text("sub/dir/note.txt", "alpha\nbeta")
            .await
            .unwrap();
        assert_eq!(
            ex.read_text("sub/dir/note.txt").await.unwrap(),
            "alpha\nbeta"
        );

        // Binary + windowed read_bytes.
        let bytes: Vec<u8> = (0u16..=255).map(|b| b as u8).collect();
        ex.write_bytes("blob.bin", &bytes).await.unwrap();
        assert_eq!(ex.read_bytes("blob.bin", None, None).await.unwrap(), bytes);
        assert_eq!(
            ex.read_bytes("blob.bin", Some(10), Some(5)).await.unwrap(),
            &bytes[10..15]
        );

        // Default-secure egress: no network unless opted in (D2/D3).
        let net = ex
            .run_command(
                "wget -T 2 -q -O- http://example.com || echo NO_NET",
                Some(30.0),
            )
            .await
            .unwrap();
        assert!(
            net.stdout.contains("NO_NET"),
            "egress must be off by default"
        );

        // The workspace mount is the host boundary: the write landed on the host.
        assert!(tmp.path().join("sub/dir/note.txt").exists());

        // C2 regression: a WINDOWED read of a missing file must error, not
        // return Ok(empty) — a pipeline's exit status is base64's, which
        // exits 0 on empty input when tail fails.
        assert!(ex
            .read_bytes("no-such-file.bin", Some(0), Some(10))
            .await
            .is_err());
        assert!(ex.read_bytes("no-such-file.bin", None, None).await.is_err());

        ex.cleanup().await;
    }
}
