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

use async_trait::async_trait;
use fabro_static::EnvVars;
use fabro_types::{CommandOutputStream, CommandTermination};
use fabro_util::time::elapsed_ms;
use tokio::io::{AsyncRead, AsyncReadExt};
use tokio::process::{Child, Command};
use tokio::sync::watch;
use tokio::task::spawn_blocking;
use tokio::{fs, time};
use tokio_util::sync::CancellationToken;

use crate::sandbox::{
    self, BASH_ENV_VAR, BASH_PROBE_SCRIPT, BASH_PROBE_TIMEOUT_MS, OutputCaptureBuffer,
    StdioProcessControl, optional_timeout, validate_bash_probe, write_process_stdin,
};
use crate::{
    CommandOutputCallback, DEFAULT_EXEC_OUTPUT_TAIL_BYTES, DirEntry, ExecResult,
    ExecStreamingRequest, ExecStreamingResult, GrepOptions, Sandbox, SandboxEvent,
    SandboxEventCallback, SandboxFile, StderrCollector, StdioProcess, StdioProcessHandle,
    StdioProcessTermination, WalkOptions,
};

/// Remediation shown when the worker has no usable Bash.
///
/// Local sandboxes resolve `bash` through `PATH` rather than requiring
/// `/bin/bash`, which is what keeps them working on NixOS.
const LOCAL_BASH_REMEDIATION: &str = "Local sandboxes require Bash on the worker PATH. Install bash, or run this workflow in a \
     Docker or Daytona sandbox.";

/// Where [`LocalSandbox`] gets its Bash executable.
///
/// The non-default variants exist so tests can exercise missing and non-Bash
/// interpreters without mutating the process-global `PATH`, which would make
/// parallel tests racy.
enum BashExecutable {
    /// Resolve `bash` through the worker's `PATH`.
    ResolveFromPath,
    #[cfg(test)]
    Fixed(PathBuf),
    #[cfg(test)]
    Unavailable,
}

pub struct LocalSandbox {
    working_directory: PathBuf,
    event_callback:    Option<SandboxEventCallback>,
    rg_available:      std::sync::OnceLock<bool>,
    bash_executable:   BashExecutable,
    bash_path:         std::sync::OnceLock<PathBuf>,
}

impl LocalSandbox {
    #[must_use]
    pub fn new(working_directory: PathBuf) -> Self {
        Self {
            working_directory,
            event_callback: None,
            rg_available: std::sync::OnceLock::new(),
            bash_executable: BashExecutable::ResolveFromPath,
            bash_path: std::sync::OnceLock::new(),
        }
    }

    #[cfg(test)]
    fn with_bash_executable(working_directory: PathBuf, bash_executable: BashExecutable) -> Self {
        Self {
            bash_executable,
            ..Self::new(working_directory)
        }
    }

    /// Resolve, and then remember, the Bash executable this sandbox runs
    /// commands with.
    ///
    /// Every command path — non-streaming, streaming, and stdio — goes through
    /// here so a single sandbox can never split across two interpreters.
    fn bash(&self) -> crate::Result<PathBuf> {
        if let Some(path) = self.bash_path.get() {
            return Ok(path.clone());
        }

        let resolved = match &self.bash_executable {
            BashExecutable::ResolveFromPath => Self::binary_path_on_path("bash")
                .ok_or_else(|| crate::Error::message(LOCAL_BASH_REMEDIATION))?,
            #[cfg(test)]
            BashExecutable::Fixed(path) => path.clone(),
            #[cfg(test)]
            BashExecutable::Unavailable => {
                return Err(crate::Error::message(LOCAL_BASH_REMEDIATION));
            }
        };

        Ok(self.bash_path.get_or_init(|| resolved).clone())
    }

    /// Verify the resolved executable is non-login Bash.
    ///
    /// Runs on fresh initialization and again on resume, so a worker that lost
    /// its Bash between runs fails before the first command instead of during
    /// it.
    async fn probe_bash(&self) -> crate::Result<()> {
        let bash = self.bash()?;
        let remediation = format!(
            "{} is not usable as non-login Bash. {LOCAL_BASH_REMEDIATION}",
            bash.display()
        );
        let result = self
            .exec_command(BASH_PROBE_SCRIPT, BASH_PROBE_TIMEOUT_MS, None, None, None)
            .await
            .map_err(|err| {
                crate::Error::context(
                    format!("Failed to run the local Bash check. {remediation}"),
                    err,
                )
            })?;
        validate_bash_probe(result, remediation)
    }

    /// Ensure command execution has a working directory and a usable Bash.
    ///
    /// Fresh initialization and resumed starts share this path so they report
    /// directory failures with the same context before attempting the Bash
    /// probe.
    async fn prepare_command_environment(&self) -> crate::Result<()> {
        fs::create_dir_all(&self.working_directory)
            .await
            .map_err(|error| crate::Error::context("Failed to create working directory", error))?;
        self.probe_bash().await
    }

    pub fn set_event_callback(&mut self, cb: SandboxEventCallback) {
        self.event_callback = Some(cb);
    }

    fn emit(&self, event: SandboxEvent) {
        event.trace();
        if let Some(ref cb) = self.event_callback {
            cb(event);
        }
    }

    const ENV_SAFELIST: &'static [&'static str] = &[
        EnvVars::PATH,
        EnvVars::HOME,
        EnvVars::USER,
        EnvVars::SHELL,
        EnvVars::LANG,
        EnvVars::TERM,
        EnvVars::TMPDIR,
        EnvVars::GOPATH,
        EnvVars::CARGO_HOME,
        EnvVars::NVM_DIR,
    ];

    fn should_filter_env_var(key: &str) -> bool {
        if Self::ENV_SAFELIST.contains(&key) {
            return false;
        }
        let lower = key.to_lowercase();
        lower.ends_with("_api_key")
            || lower.ends_with("_secret")
            || lower.ends_with("_token")
            || lower.ends_with("_password")
            || lower.ends_with("_credential")
    }

    fn resolve_path(&self, path: &str) -> PathBuf {
        let p = Path::new(path);
        if p.is_absolute() {
            p.to_path_buf()
        } else {
            self.working_directory.join(p)
        }
    }

    fn binary_on_path(binary: &str) -> bool {
        Self::binary_path_on_path(binary).is_some()
    }

    #[expect(
        clippy::disallowed_methods,
        reason = "Local sandbox command execution checks PATH/PATHEXT to select optional helpers."
    )]
    fn binary_path_on_path(binary: &str) -> Option<PathBuf> {
        let paths = std::env::var_os(EnvVars::PATH)?;

        #[cfg(windows)]
        let extensions: Vec<String> = std::env::var_os(EnvVars::PATHEXT)
            .map(|value| {
                value
                    .to_string_lossy()
                    .split(';')
                    .map(|ext| ext.to_ascii_lowercase())
                    .collect()
            })
            .unwrap_or_else(|| vec![".exe".to_string(), ".cmd".to_string(), ".bat".to_string()]);

        for dir in std::env::split_paths(&paths) {
            let candidate = dir.join(binary);
            if candidate.is_file() {
                return Some(candidate);
            }

            #[cfg(windows)]
            {
                if candidate.extension().is_none() {
                    for ext in &extensions {
                        let with_ext = dir.join(format!("{binary}{ext}"));
                        if with_ext.is_file() {
                            return Some(with_ext);
                        }
                    }
                }
            }
        }

        None
    }
}

#[expect(
    clippy::disallowed_methods,
    reason = "Local sandbox must snapshot the ambient process env before applying its fail-closed filter."
)]
fn process_env_vars() -> Vec<(String, String)> {
    std::env::vars().collect()
}

#[derive(Debug, Clone, Copy)]
enum ExplicitEnvPolicy {
    FilterSensitive,
    TrustCaller,
}

fn filtered_env_vars(
    env_vars: Option<&std::collections::HashMap<String, String>>,
    explicit_policy: ExplicitEnvPolicy,
) -> Vec<(String, String)> {
    let mut filtered_env: Vec<(String, String)> = process_env_vars()
        .into_iter()
        .filter(|(key, _)| key != BASH_ENV_VAR && !LocalSandbox::should_filter_env_var(key))
        .collect();

    if let Some(extra) = env_vars {
        for (key, value) in extra {
            if key != BASH_ENV_VAR
                && (matches!(explicit_policy, ExplicitEnvPolicy::TrustCaller)
                    || !LocalSandbox::should_filter_env_var(key))
            {
                filtered_env.push((key.clone(), value.clone()));
            }
        }
    }

    filtered_env
}

async fn drain_pipe<R>(mut pipe: Option<R>, stream: CommandOutputStream) -> String
where
    R: AsyncRead + Unpin,
{
    let mut buf = String::new();
    if let Some(ref mut reader) = pipe {
        if let Err(err) = reader.read_to_string(&mut buf).await {
            tracing::warn!(error = %err, ?stream, "Failed to drain child output");
        }
    }
    buf
}

type LocalStdioOutcome = Result<StdioProcessTermination, String>;

struct LocalStdioProcessControl {
    terminate_tx:   watch::Sender<bool>,
    termination_rx: watch::Receiver<Option<LocalStdioOutcome>>,
}

impl LocalStdioProcessControl {
    fn new(mut child: Child) -> Self {
        let (terminate_tx, mut terminate_rx) = watch::channel(false);
        let (termination_tx, termination_rx) = watch::channel(None);

        tokio::spawn(async move {
            let outcome = tokio::select! {
                status = child.wait() => {
                    status
                        .map(|status| StdioProcessTermination::exited(status.code()))
                        .map_err(|err| format!("Failed to wait for stdio process: {err}"))
                }
                changed = terminate_rx.changed() => {
                    if changed.is_err() || !*terminate_rx.borrow() {
                        child.wait()
                            .await
                            .map(|status| StdioProcessTermination::exited(status.code()))
                            .map_err(|err| format!("Failed to wait for stdio process: {err}"))
                    } else {
                        sigterm_then_kill(&mut child).await;
                        Ok(StdioProcessTermination::cancelled())
                    }
                }
            };
            let _ = termination_tx.send(Some(outcome));
        });

        Self {
            terminate_tx,
            termination_rx,
        }
    }

    async fn wait_for_termination(&self) -> crate::Result<StdioProcessTermination> {
        let mut termination_rx = self.termination_rx.clone();
        loop {
            if let Some(outcome) = termination_rx.borrow().clone() {
                return outcome.map_err(crate::Error::message);
            }
            termination_rx.changed().await.map_err(|_| {
                crate::Error::message(
                    "stdio process supervisor stopped before reporting termination",
                )
            })?;
        }
    }
}

#[async_trait]
impl StdioProcessControl for LocalStdioProcessControl {
    async fn terminate(&self) -> crate::Result<()> {
        if self.termination_rx.borrow().is_some() {
            return Ok(());
        }

        self.terminate_tx.send_replace(true);
        self.wait_for_termination().await.map(|_| ())
    }

    async fn wait(&self) -> crate::Result<StdioProcessTermination> {
        self.wait_for_termination().await
    }
}

#[async_trait]
impl Sandbox for LocalSandbox {
    async fn read_file_bytes(&self, path: &str) -> crate::Result<Vec<u8>> {
        let full_path = self.resolve_path(path);
        fs::read(&full_path).await.map_err(|e| {
            crate::Error::context(format!("Failed to read {}", full_path.display()), e)
        })
    }

    async fn write_file(&self, path: &str, content: &str) -> crate::Result<()> {
        let full_path = self.resolve_path(path);
        if let Some(parent) = full_path.parent() {
            fs::create_dir_all(parent)
                .await
                .map_err(|e| crate::Error::context("Failed to create parent dirs", e))?;
        }
        fs::write(&full_path, content).await.map_err(|e| {
            crate::Error::context(format!("Failed to write {}", full_path.display()), e)
        })
    }

    async fn delete_file(&self, path: &str) -> crate::Result<()> {
        let full_path = self.resolve_path(path);
        fs::remove_file(&full_path).await.map_err(|e| {
            crate::Error::context(format!("Failed to delete {}", full_path.display()), e)
        })
    }

    async fn file_exists(&self, path: &str) -> crate::Result<bool> {
        let full_path = self.resolve_path(path);
        Ok(full_path.exists())
    }

    async fn list_directory(
        &self,
        path: &str,
        depth: Option<usize>,
    ) -> crate::Result<Vec<DirEntry>> {
        #[expect(
            clippy::disallowed_methods,
            reason = "sync recursive read_dir; caller wraps invocation in tokio::task::spawn_blocking"
        )]
        fn list_recursive(
            base: &std::path::Path,
            prefix: &str,
            current_depth: usize,
            max_depth: usize,
            entries: &mut Vec<DirEntry>,
        ) -> crate::Result<()> {
            let mut dir_entries: Vec<std::fs::DirEntry> = std::fs::read_dir(base)
                .map_err(|e| {
                    crate::Error::context(format!("Failed to read directory {}", base.display()), e)
                })?
                .filter_map(std::result::Result::ok)
                .collect();
            dir_entries.sort_by_key(std::fs::DirEntry::file_name);

            for entry in dir_entries {
                let metadata = entry
                    .metadata()
                    .map_err(|e| crate::Error::context("Failed to read metadata", e))?;
                let name = if prefix.is_empty() {
                    entry.file_name().to_string_lossy().into_owned()
                } else {
                    format!("{prefix}/{}", entry.file_name().to_string_lossy())
                };
                let is_dir = metadata.is_dir();
                entries.push(DirEntry {
                    name: name.clone(),
                    is_dir,
                    size: if metadata.is_file() {
                        Some(metadata.len())
                    } else {
                        None
                    },
                });
                if is_dir && current_depth + 1 < max_depth {
                    list_recursive(&entry.path(), &name, current_depth + 1, max_depth, entries)?;
                }
            }
            Ok(())
        }

        let full_path = self.resolve_path(path);
        let max_depth = depth.unwrap_or(1);
        spawn_blocking(move || {
            let mut entries = Vec::new();
            list_recursive(&full_path, "", 0, max_depth, &mut entries)?;
            Ok(entries)
        })
        .await
        .map_err(|e| crate::Error::context("list_directory task failed", e))?
    }

    async fn exec_command(
        &self,
        command: &str,
        timeout_ms: u64,
        working_dir: Option<&str>,
        env_vars: Option<&std::collections::HashMap<String, String>>,
        cancel_token: Option<CancellationToken>,
    ) -> crate::Result<ExecResult> {
        let start = Instant::now();

        let filtered_env = filtered_env_vars(env_vars, ExplicitEnvPolicy::FilterSensitive);

        let effective_dir =
            working_dir.map_or_else(|| self.working_directory.clone(), std::path::PathBuf::from);

        let mut cmd = Command::new(self.bash()?);
        cmd.arg("-c")
            .arg(command)
            .current_dir(&effective_dir)
            .env_clear()
            .envs(filtered_env)
            .kill_on_drop(true)
            .stdout(std::process::Stdio::piped())
            .stderr(std::process::Stdio::piped());

        #[cfg(unix)]
        fabro_proc::pre_exec_setpgid(cmd.as_std_mut());

        let mut child = cmd
            .spawn()
            .map_err(|e| crate::Error::context("Failed to spawn command", e))?;

        let timeout_duration = std::time::Duration::from_millis(timeout_ms);
        let token = cancel_token.unwrap_or_default();

        // Take stdout/stderr handles before entering the select! so we can
        // drain them concurrently.  Without this, the child can deadlock: if
        // it writes more than the OS pipe buffer (~64 KB) the write() syscall
        // blocks until the parent drains the pipe, but the parent is blocked
        // on child.wait().
        let stdout_pipe = child.stdout.take();
        let stderr_pipe = child.stderr.take();
        let stdout_task =
            tokio::spawn(async move { drain_pipe(stdout_pipe, CommandOutputStream::Stdout).await });
        let stderr_task =
            tokio::spawn(async move { drain_pipe(stderr_pipe, CommandOutputStream::Stderr).await });

        let (termination, exit_code) = tokio::select! {
            status_result = child.wait() => {
                let status = status_result
                    .map_err(|e| crate::Error::context("Failed to wait for process", e))?;
                (CommandTermination::Exited, status.code())
            }
            () = time::sleep(timeout_duration) => {
                sigterm_then_kill(&mut child).await;
                (CommandTermination::TimedOut, None)
            }
            () = token.cancelled() => {
                sigterm_then_kill(&mut child).await;
                (CommandTermination::Cancelled, None)
            }
        };

        let duration_ms = elapsed_ms(start);

        let stdout_str = stdout_task.await.unwrap_or_default();
        let stderr_str = stderr_task.await.unwrap_or_default();

        Ok(ExecResult {
            stdout: stdout_str,
            stderr: stderr_str,
            exit_code,
            termination,
            duration_ms,
        })
    }

    async fn exec_command_streaming(
        &self,
        request: ExecStreamingRequest<'_>,
    ) -> crate::Result<ExecStreamingResult> {
        let ExecStreamingRequest {
            command,
            timeout_ms,
            working_dir,
            env_vars,
            cancel_token,
            stdin,
            output_callback,
            stream_output_bytes_cap,
        } = request;
        let start = Instant::now();

        let filtered_env = filtered_env_vars(env_vars, ExplicitEnvPolicy::FilterSensitive);

        let effective_dir =
            working_dir.map_or_else(|| self.working_directory.clone(), std::path::PathBuf::from);

        let mut cmd = Command::new(self.bash()?);
        cmd.arg("-c")
            .arg(command)
            .current_dir(&effective_dir)
            .env_clear()
            .envs(filtered_env)
            .kill_on_drop(true)
            .stdout(std::process::Stdio::piped())
            .stderr(std::process::Stdio::piped());
        if stdin.is_some() {
            cmd.stdin(std::process::Stdio::piped());
        }

        #[cfg(unix)]
        fabro_proc::pre_exec_setpgid(cmd.as_std_mut());

        let mut child = cmd
            .spawn()
            .map_err(|e| crate::Error::context("Failed to spawn command", e))?;

        let timeout_future = optional_timeout(timeout_ms);
        tokio::pin!(timeout_future);
        let token = cancel_token.unwrap_or_default();

        let stdout_pipe = child.stdout.take();
        let stderr_pipe = child.stderr.take();
        let stdin_task = match stdin {
            Some(stdin) => {
                let stdin_pipe = child.stdin.take().ok_or_else(|| {
                    crate::Error::message("Failed to open command standard input")
                })?;
                Some(tokio::spawn(async move {
                    write_process_stdin(stdin_pipe, &stdin).await
                }))
            }
            None => None,
        };
        let stdout_callback = output_callback.clone();
        let stderr_callback = output_callback;
        let stdout_task = tokio::spawn(async move {
            drain_command_pipe(
                stdout_pipe,
                CommandOutputStream::Stdout,
                stdout_callback,
                stream_output_bytes_cap,
            )
            .await
        });
        let stderr_task = tokio::spawn(async move {
            drain_command_pipe(
                stderr_pipe,
                CommandOutputStream::Stderr,
                stderr_callback,
                stream_output_bytes_cap,
            )
            .await
        });

        let (termination, exit_code) = tokio::select! {
            status_result = child.wait() => {
                let status = status_result
                    .map_err(|e| crate::Error::context("Failed to wait for process", e))?;
                (CommandTermination::Exited, status.code())
            }
            () = &mut timeout_future => {
                sigterm_then_kill(&mut child).await;
                (CommandTermination::TimedOut, None)
            }
            () = token.cancelled() => {
                sigterm_then_kill(&mut child).await;
                (CommandTermination::Cancelled, None)
            }
        };

        let duration_ms = elapsed_ms(start);
        if let Some(stdin_task) = stdin_task {
            // The process is gone, so unwritten stdin bytes are unwanted.
            // Abort instead of joining unbounded: a backgrounded grandchild
            // that inherited the pipe could otherwise block the writer
            // forever.
            stdin_task.abort();
            match stdin_task.await {
                Ok(result) => result?,
                Err(join_error) if join_error.is_cancelled() => {}
                Err(join_error) => {
                    return Err(crate::Error::context(
                        "stdin stream task failed",
                        join_error,
                    ));
                }
            }
        }
        let stdout_capture = stdout_task
            .await
            .map_err(|e| crate::Error::context("stdout stream task failed", e))??;
        let stderr_capture = stderr_task
            .await
            .map_err(|e| crate::Error::context("stderr stream task failed", e))??;
        let (stdout_bytes, stdout_capture) = stdout_capture.into_parts();
        let (stderr_bytes, stderr_capture) = stderr_capture.into_parts();

        Ok(ExecStreamingResult {
            result: ExecResult {
                stdout: String::from_utf8_lossy(&stdout_bytes).into_owned(),
                stderr: String::from_utf8_lossy(&stderr_bytes).into_owned(),
                exit_code,
                termination,
                duration_ms,
            },
            streams_separated: true,
            live_streaming: true,
            stdout_capture,
            stderr_capture,
        })
    }

    async fn spawn_stdio_process(
        &self,
        command: &str,
        working_dir: Option<&str>,
        env_vars: Option<&std::collections::HashMap<String, String>>,
        cancel_token: Option<CancellationToken>,
    ) -> crate::Result<StdioProcess> {
        let filtered_env = filtered_env_vars(env_vars, ExplicitEnvPolicy::TrustCaller);

        let effective_dir =
            working_dir.map_or_else(|| self.working_directory.clone(), std::path::PathBuf::from);

        let mut cmd = Command::new(self.bash()?);
        cmd.arg("-c")
            .arg(format!("exec {command}"))
            .current_dir(&effective_dir)
            .env_clear()
            .envs(filtered_env)
            .stdin(std::process::Stdio::piped())
            .stdout(std::process::Stdio::piped())
            .stderr(std::process::Stdio::piped());

        #[cfg(unix)]
        fabro_proc::pre_exec_setpgid(cmd.as_std_mut());

        let mut child = cmd
            .spawn()
            .map_err(|e| crate::Error::context("Failed to spawn stdio process", e))?;

        let stdin = child
            .stdin
            .take()
            .ok_or_else(|| crate::Error::message("Failed to open stdio process stdin"))?;
        let stdout = child
            .stdout
            .take()
            .ok_or_else(|| crate::Error::message("Failed to open stdio process stdout"))?;
        let stderr = child
            .stderr
            .take()
            .ok_or_else(|| crate::Error::message("Failed to open stdio process stderr"))?;

        let stderr_collector = StderrCollector::new(DEFAULT_EXEC_OUTPUT_TAIL_BYTES);
        stderr_collector.spawn_reader(stderr);

        let handle = StdioProcessHandle::new(LocalStdioProcessControl::new(child));

        if let Some(token) = cancel_token {
            let handle_for_cancel = handle.clone();
            tokio::spawn(async move {
                token.cancelled().await;
                if let Err(err) = handle_for_cancel.terminate().await {
                    tracing::warn!(error = %err, "Failed to terminate cancelled stdio process");
                }
            });
        }

        Ok(StdioProcess {
            stdin: Box::pin(stdin),
            stdout: Box::pin(stdout),
            stderr: stderr_collector,
            handle,
        })
    }

    async fn grep(
        &self,
        pattern: &str,
        path: &str,
        options: &GrepOptions,
    ) -> crate::Result<Vec<String>> {
        let full_path = self.resolve_path(path);

        // Try rg (ripgrep) first, fall back to grep
        let use_rg = *self.rg_available.get_or_init(|| Self::binary_on_path("rg"));

        let output = if use_rg {
            let mut args = vec!["-n".to_string()];
            if options.case_insensitive {
                args.push("-i".into());
            }
            if let Some(ref glob_filter) = options.glob_filter {
                args.push("--glob".into());
                args.push(glob_filter.clone());
            }
            if let Some(max) = options.max_results {
                args.push("-m".into());
                args.push(max.to_string());
            }
            args.push(pattern.into());
            args.push(full_path.to_string_lossy().into_owned());

            Command::new("rg")
                .args(&args)
                .output()
                .await
                .map_err(|e| crate::Error::context("Failed to run rg", e))?
        } else {
            let mut args = vec!["-rn".to_string()];
            if options.case_insensitive {
                args.push("-i".into());
            }
            if let Some(ref glob_filter) = options.glob_filter {
                args.push("--include".into());
                args.push(glob_filter.clone());
            }
            if let Some(max) = options.max_results {
                args.push("-m".into());
                args.push(max.to_string());
            }
            args.push(pattern.into());
            args.push(full_path.to_string_lossy().into_owned());

            Command::new("grep")
                .args(&args)
                .output()
                .await
                .map_err(|e| crate::Error::context("Failed to run grep", e))?
        };

        let stdout = String::from_utf8_lossy(&output.stdout);
        let results: Vec<String> = stdout
            .lines()
            .map(String::from)
            .filter(|l| !l.is_empty())
            .collect();
        Ok(results)
    }

    async fn walk_files(
        &self,
        base: &str,
        relative_start: &str,
        options: &WalkOptions,
    ) -> crate::Result<Vec<SandboxFile>> {
        let base = self.resolve_path(base);
        walk_local_files(&base, relative_start, options).await
    }

    async fn download_file_to_local(
        &self,
        remote_path: &str,
        local_path: &Path,
    ) -> crate::Result<()> {
        let full_path = self.resolve_path(remote_path);
        if let Some(parent) = local_path.parent() {
            fs::create_dir_all(parent)
                .await
                .map_err(|e| crate::Error::context("Failed to create parent dirs", e))?;
        }
        fs::copy(&full_path, local_path).await.map_err(|e| {
            crate::Error::context(
                format!(
                    "Failed to copy {} to {}",
                    full_path.display(),
                    local_path.display()
                ),
                e,
            )
        })?;
        Ok(())
    }

    async fn upload_file_from_local(
        &self,
        local_path: &Path,
        remote_path: &str,
    ) -> crate::Result<()> {
        let full_path = self.resolve_path(remote_path);
        if let Some(parent) = full_path.parent() {
            fs::create_dir_all(parent)
                .await
                .map_err(|e| crate::Error::context("Failed to create parent dirs", e))?;
        }
        fs::copy(local_path, &full_path).await.map_err(|e| {
            crate::Error::context(
                format!(
                    "Failed to copy {} to {}",
                    local_path.display(),
                    full_path.display()
                ),
                e,
            )
        })?;
        Ok(())
    }

    async fn initialize(&self) -> crate::Result<()> {
        self.emit(SandboxEvent::Initializing {
            provider: "local".into(),
        });
        let start = Instant::now();
        let result = self.prepare_command_environment().await;
        let duration_ms = u64::try_from(start.elapsed().as_millis()).unwrap_or(u64::MAX);
        match &result {
            Ok(()) => self.emit(SandboxEvent::Ready {
                provider: "local".into(),
                duration_ms,
                name: None,
                cpu: None,
                memory: None,
                url: None,
            }),
            Err(e) => self.emit(SandboxEvent::InitializeFailed {
                provider: "local".into(),
                error: e.to_string(),
                causes: e.causes(),
                duration_ms,
            }),
        }
        result
    }

    /// Re-verify Bash before a resumed workflow can issue commands.
    ///
    /// A worker that was rebuilt or reprovisioned between runs can lose the
    /// Bash it initialized with; there is deliberately no fallback interpreter
    /// when that happens.
    async fn start(&self) -> crate::Result<()> {
        self.emit(SandboxEvent::StartStarted {
            provider: "local".into(),
        });
        let start = Instant::now();
        let result = self.prepare_command_environment().await;
        match &result {
            Ok(()) => self.emit(SandboxEvent::StartCompleted {
                provider:    "local".into(),
                duration_ms: elapsed_ms(start),
            }),
            Err(err) => self.emit(SandboxEvent::StartFailed {
                provider: "local".into(),
                error:    err.to_string(),
                causes:   err.causes(),
            }),
        }
        result
    }

    async fn activate(&self) -> crate::Result<()> {
        // Local sandboxes have no provider resource that can stop or pause.
        // Resume paths still call `start()` to recreate the directory and
        // verify Bash.
        Ok(())
    }

    async fn git_push_ref(
        &self,
        refspec: &str,
        plan: &crate::RetryPlan,
    ) -> Result<crate::PushReport, crate::PushError> {
        let has_origin = match self
            .exec_command("git remote get-url origin", 10_000, None, None, None)
            .await
        {
            Ok(result) if result.is_success() => true,
            Ok(_) => false,
            Err(err) => {
                return Err(crate::PushError {
                    report: crate::PushReport::default(),
                    error:  crate::Error::context("git remote get-url origin", err),
                });
            }
        };
        if !has_origin {
            return Ok(crate::PushReport::default());
        }

        // Local pushes use whatever credentials the host repository already
        // carries; there is no managed credential state to lease.
        sandbox::git_push_via_exec(self, None, refspec, plan).await
    }

    async fn cleanup(&self) -> crate::Result<()> {
        self.emit(SandboxEvent::CleanupStarted {
            provider: "local".into(),
        });
        let start = Instant::now();
        let duration_ms = u64::try_from(start.elapsed().as_millis()).unwrap_or(u64::MAX);
        self.emit(SandboxEvent::CleanupCompleted {
            provider: "local".into(),
            duration_ms,
        });
        Ok(())
    }

    async fn stop(&self) -> crate::Result<()> {
        self.emit(SandboxEvent::StopStarted {
            provider: "local".into(),
        });
        let start = Instant::now();
        let duration_ms = u64::try_from(start.elapsed().as_millis()).unwrap_or(u64::MAX);
        self.emit(SandboxEvent::StopCompleted {
            provider: "local".into(),
            duration_ms,
        });
        Ok(())
    }

    async fn delete(&self) -> crate::Result<()> {
        Ok(())
    }

    fn working_directory(&self) -> &str {
        self.working_directory.to_str().unwrap_or(".")
    }

    fn platform(&self) -> &str {
        if cfg!(target_os = "macos") {
            "darwin"
        } else if cfg!(target_os = "linux") {
            "linux"
        } else if cfg!(target_os = "windows") {
            "windows"
        } else {
            "unknown"
        }
    }

    #[expect(
        clippy::disallowed_methods,
        reason = "This synchronous host metadata probe only runs uname once while building the sandbox platform string."
    )]
    fn os_version(&self) -> String {
        #[cfg(unix)]
        {
            let output = std::process::Command::new("uname").arg("-r").output();
            match output {
                Ok(out) => {
                    let version = String::from_utf8_lossy(&out.stdout).trim().to_string();
                    format!("{} {version}", self.platform())
                }
                Err(_) => self.platform().to_string(),
            }
        }
        #[cfg(not(unix))]
        {
            self.platform().to_string()
        }
    }
}

/// Send SIGTERM to the process group, wait 2s for graceful shutdown, then
/// SIGKILL.
async fn sigterm_then_kill(child: &mut Child) {
    #[cfg(unix)]
    if let Some(pid) = child.id() {
        fabro_proc::sigterm_process_group(pid);
        if time::timeout(std::time::Duration::from_secs(2), child.wait())
            .await
            .is_err()
        {
            let _ = child.kill().await;
            let _ = child.wait().await;
        }
    } else {
        let _ = child.kill().await;
        let _ = child.wait().await;
    }
    #[cfg(not(unix))]
    {
        let _ = child.kill().await;
        let _ = child.wait().await;
    }
}

async fn drain_command_pipe<R>(
    mut reader: Option<R>,
    stream: CommandOutputStream,
    output_callback: Option<CommandOutputCallback>,
    stream_output_bytes_cap: Option<usize>,
) -> crate::Result<OutputCaptureBuffer>
where
    R: AsyncRead + Unpin,
{
    let mut output = OutputCaptureBuffer::new(stream_output_bytes_cap);
    let Some(reader) = reader.as_mut() else {
        return Ok(output);
    };

    let mut buf = [0_u8; 8192];
    loop {
        let read = reader
            .read(&mut buf)
            .await
            .map_err(|e| crate::Error::context("Failed to read command output", e))?;
        if read == 0 {
            return Ok(output);
        }
        output.push(&buf[..read]);
        if let Some(output_callback) = output_callback.as_ref() {
            output_callback(stream, buf[..read].to_vec()).await?;
        }
    }
}

async fn walk_local_files(
    base: &Path,
    relative_start: &str,
    options: &WalkOptions,
) -> crate::Result<Vec<SandboxFile>> {
    if options.excludes_relative_path(relative_start) {
        return Ok(Vec::new());
    }

    let Some((root, root_metadata)) = resolve_local_walk_root(base, relative_start).await? else {
        return Ok(Vec::new());
    };
    let mut files = Vec::new();
    let mut stack = vec![(root, Some(root_metadata))];

    while let Some((path, known_metadata)) = stack.pop() {
        let metadata = match known_metadata {
            Some(metadata) => metadata,
            None => match fs::symlink_metadata(&path).await {
                Ok(metadata) => metadata,
                Err(err) if err.kind() == std::io::ErrorKind::NotFound => continue,
                Err(err) => {
                    return Err(crate::Error::context(
                        format!("Failed to stat {}", path.display()),
                        err,
                    ));
                }
            },
        };

        let file_type = metadata.file_type();
        if file_type.is_file() {
            let relative_path = path.strip_prefix(base).map_err(|error| {
                crate::Error::context(
                    format!(
                        "Failed to make {} relative to {}",
                        path.display(),
                        base.display()
                    ),
                    error,
                )
            })?;
            files.push(SandboxFile {
                path:          path.to_string_lossy().into_owned(),
                relative_path: relative_path
                    .to_string_lossy()
                    .replace(std::path::MAIN_SEPARATOR, "/"),
                size:          metadata.len(),
            });
        } else if file_type.is_dir() {
            if path != base
                && path
                    .file_name()
                    .and_then(std::ffi::OsStr::to_str)
                    .is_some_and(|file_name| options.excludes_name(file_name))
            {
                continue;
            }
            let mut entries = match fs::read_dir(&path).await {
                Ok(entries) => entries,
                Err(err) if err.kind() == std::io::ErrorKind::NotFound => continue,
                Err(err) => {
                    return Err(crate::Error::context(
                        format!("Failed to read directory {}", path.display()),
                        err,
                    ));
                }
            };
            while let Some(entry) = entries.next_entry().await.map_err(|err| {
                crate::Error::context(
                    format!("Failed to read directory entry in {}", path.display()),
                    err,
                )
            })? {
                stack.push((entry.path(), None));
            }
        }
    }

    Ok(files)
}

async fn resolve_local_walk_root(
    base: &Path,
    relative_start: &str,
) -> crate::Result<Option<(PathBuf, std::fs::Metadata)>> {
    if relative_start.is_empty() {
        return match fs::metadata(base).await {
            Ok(metadata) => Ok(Some((base.to_path_buf(), metadata))),
            Err(error)
                if matches!(
                    error.kind(),
                    std::io::ErrorKind::NotFound | std::io::ErrorKind::NotADirectory
                ) =>
            {
                Ok(None)
            }
            Err(error) => Err(crate::Error::context(
                format!("Failed to stat traversal base {}", base.display()),
                error,
            )),
        };
    }

    let mut root = base.to_path_buf();
    let mut metadata = None;
    for segment in relative_start.split('/') {
        root.push(segment);
        let component_metadata = match fs::symlink_metadata(&root).await {
            Ok(metadata) => metadata,
            Err(error)
                if matches!(
                    error.kind(),
                    std::io::ErrorKind::NotFound | std::io::ErrorKind::NotADirectory
                ) =>
            {
                return Ok(None);
            }
            Err(error) => {
                return Err(crate::Error::context(
                    format!("Failed to stat traversal root component {}", root.display()),
                    error,
                ));
            }
        };
        if component_metadata.file_type().is_symlink() {
            return Ok(None);
        }
        metadata = Some(component_metadata);
    }

    Ok(metadata.map(|metadata| (root, metadata)))
}

#[cfg(test)]
#[expect(
    clippy::disallowed_methods,
    reason = "sandbox tests stage fixtures with sync std::fs writes/reads"
)]
mod tests {
    use std::collections::HashMap;
    use std::io;
    use std::path::PathBuf;
    use std::pin::Pin;
    use std::sync::{Arc, Mutex};
    use std::task::{Context as TaskContext, Poll};

    use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader, ReadBuf};

    use super::*;

    fn temp_dir() -> PathBuf {
        let dir = std::env::temp_dir().join(format!("local_env_test_{}", uuid::Uuid::new_v4()));
        std::fs::create_dir_all(&dir).unwrap();
        dir
    }

    #[tokio::test]
    async fn drain_pipe_returns_empty_buffer_after_read_failure() {
        struct FailingReader;

        impl AsyncRead for FailingReader {
            fn poll_read(
                self: Pin<&mut Self>,
                _cx: &mut TaskContext<'_>,
                _buf: &mut ReadBuf<'_>,
            ) -> Poll<io::Result<()>> {
                Poll::Ready(Err(io::Error::other("simulated read failure")))
            }
        }

        let output = drain_pipe(Some(FailingReader), CommandOutputStream::Stdout).await;

        assert!(output.is_empty());
    }

    #[tokio::test]
    async fn read_file_with_line_numbers() {
        let dir = temp_dir();
        std::fs::write(dir.join("test.txt"), "hello\nworld\nfoo").unwrap();

        let env = LocalSandbox::new(dir.clone());
        let result = env.read_file("test.txt", None, None).await.unwrap();

        assert_eq!(result, "1 | hello\n2 | world\n3 | foo\n");
        std::fs::remove_dir_all(&dir).unwrap();
    }

    #[tokio::test]
    async fn read_file_line_number_padding() {
        let dir = temp_dir();
        let content =
            (1..=12)
                .map(|i| format!("line {i}\n"))
                .fold(String::new(), |mut acc, line| {
                    acc.push_str(&line);
                    acc
                });
        std::fs::write(dir.join("padded.txt"), content.trim_end()).unwrap();

        let env = LocalSandbox::new(dir.clone());
        let result = env.read_file("padded.txt", None, None).await.unwrap();

        assert!(result.starts_with(" 1 | line 1\n"));
        assert!(result.contains("12 | line 12\n"));
        std::fs::remove_dir_all(&dir).unwrap();
    }

    #[tokio::test]
    async fn read_file_not_found() {
        let dir = temp_dir();
        let env = LocalSandbox::new(dir.clone());
        let result = env.read_file("nonexistent.txt", None, None).await;
        assert!(result.is_err());
        std::fs::remove_dir_all(&dir).unwrap();
    }

    #[tokio::test]
    async fn write_file_creates_parent_dirs() {
        let dir = temp_dir();
        let env = LocalSandbox::new(dir.clone());
        env.write_file("sub/dir/test.txt", "content").await.unwrap();

        let written = std::fs::read_to_string(dir.join("sub/dir/test.txt")).unwrap();
        assert_eq!(written, "content");
        std::fs::remove_dir_all(&dir).unwrap();
    }

    #[tokio::test]
    async fn file_exists_true() {
        let dir = temp_dir();
        std::fs::write(dir.join("exists.txt"), "data").unwrap();

        let env = LocalSandbox::new(dir.clone());
        assert!(env.file_exists("exists.txt").await.unwrap());
        std::fs::remove_dir_all(&dir).unwrap();
    }

    #[tokio::test]
    async fn file_exists_false() {
        let dir = temp_dir();
        let env = LocalSandbox::new(dir.clone());
        assert!(!env.file_exists("nope.txt").await.unwrap());
        std::fs::remove_dir_all(&dir).unwrap();
    }

    #[tokio::test]
    async fn list_directory_sorted() {
        let dir = temp_dir();
        std::fs::write(dir.join("b.txt"), "b").unwrap();
        std::fs::write(dir.join("a.txt"), "a").unwrap();
        std::fs::create_dir(dir.join("c_dir")).unwrap();

        let env = LocalSandbox::new(dir.clone());
        let entries = env.list_directory(".", None).await.unwrap();

        assert_eq!(entries.len(), 3);
        assert_eq!(entries[0].name, "a.txt");
        assert!(!entries[0].is_dir);
        assert!(entries[0].size.is_some());
        assert_eq!(entries[1].name, "b.txt");
        assert_eq!(entries[2].name, "c_dir");
        assert!(entries[2].is_dir);
        assert!(entries[2].size.is_none());
        std::fs::remove_dir_all(&dir).unwrap();
    }

    #[tokio::test]
    async fn exec_command_echo() {
        let dir = temp_dir();
        let env = LocalSandbox::new(dir.clone());
        let result = env
            .exec_command("echo hello", 5000, None, None, None)
            .await
            .unwrap();

        assert_eq!(result.stdout.trim(), "hello");
        assert_eq!(result.exit_code, Some(0));
        assert_eq!(result.termination, CommandTermination::Exited);
        assert!(result.duration_ms < 5000);
        std::fs::remove_dir_all(&dir).unwrap();
    }

    #[tokio::test]
    async fn stdio_process_round_trips_lines() {
        let dir = temp_dir();
        let sandbox = LocalSandbox::new(dir.clone());
        let process = sandbox
            .spawn_stdio_process(
                "python3 -u -c 'import sys; [print(line.strip()[::-1], flush=True) for line in sys.stdin]'",
                None,
                None,
                None,
            )
            .await
            .unwrap();

        let mut stdin = process.stdin;
        let mut stdout = BufReader::new(process.stdout);

        stdin.write_all(b"abc\n").await.unwrap();
        stdin.flush().await.unwrap();

        let mut line = String::new();
        stdout.read_line(&mut line).await.unwrap();
        assert_eq!(line.trim_end(), "cba");

        process.handle.terminate().await.unwrap();
        std::fs::remove_dir_all(&dir).unwrap();
    }

    #[tokio::test]
    async fn stdio_process_forwards_explicit_provider_credentials() {
        let dir = temp_dir();
        let sandbox = LocalSandbox::new(dir.clone());
        let env = HashMap::from([("OPENAI_API_KEY".to_string(), "test-key".to_string())]);
        let process = sandbox
            .spawn_stdio_process(
                "python3 -u -c 'import os; print(os.environ.get(\"OPENAI_API_KEY\", \"missing\"), flush=True)'",
                None,
                Some(&env),
                None,
            )
            .await
            .unwrap();

        let mut stdout = BufReader::new(process.stdout);
        let mut line = String::new();
        stdout.read_line(&mut line).await.unwrap();
        assert_eq!(line.trim_end(), "test-key");

        process.handle.wait().await.unwrap();
        std::fs::remove_dir_all(&dir).unwrap();
    }

    /// Bash-only syntax: `[[ ]]`, arrays, and `${arr[@]}` all fail under `sh`.
    const BASH_ONLY_COMMAND: &str = "arr=(one two three); [[ ${#arr[@]} -eq 3 ]] && echo \
                                     ${arr[1]}";

    /// Prints `login` or `nonlogin` for the shell evaluating it.
    const LOGIN_SHELL_REPORT: &str = "shopt -q login_shell && echo login || echo nonlogin";

    #[tokio::test]
    async fn exec_command_runs_bash_only_syntax() {
        let dir = temp_dir();
        let sandbox = LocalSandbox::new(dir.clone());

        let result = sandbox
            .exec_command(BASH_ONLY_COMMAND, 5000, None, None, None)
            .await
            .unwrap();

        assert_eq!(result.exit_code, Some(0), "stderr: {}", result.stderr);
        assert_eq!(result.stdout.trim(), "two");
        std::fs::remove_dir_all(&dir).unwrap();
    }

    #[tokio::test]
    async fn exec_command_streaming_runs_bash_only_syntax() {
        let dir = temp_dir();
        let sandbox = LocalSandbox::new(dir.clone());

        let result = sandbox
            .exec_command_streaming(ExecStreamingRequest {
                timeout_ms: Some(5000),
                output_callback: Some(Arc::new(|_, _| Box::pin(async { Ok(()) }))),
                ..ExecStreamingRequest::new(BASH_ONLY_COMMAND)
            })
            .await
            .unwrap();

        assert_eq!(
            result.result.exit_code,
            Some(0),
            "stderr: {}",
            result.result.stderr
        );
        assert_eq!(result.result.stdout.trim(), "two");
        std::fs::remove_dir_all(&dir).unwrap();
    }

    #[tokio::test]
    async fn exec_command_streaming_drains_full_output_after_capture_cap() {
        let dir = temp_dir();
        let sandbox = LocalSandbox::new(dir.clone());
        let callback_bytes = Arc::new(Mutex::new(Vec::new()));
        let callback_bytes_for_stream = Arc::clone(&callback_bytes);

        let result = sandbox
            .exec_command_streaming(ExecStreamingRequest {
                timeout_ms: Some(5000),
                output_callback: Some(Arc::new(move |stream, bytes| {
                    let callback_bytes = Arc::clone(&callback_bytes_for_stream);
                    Box::pin(async move {
                        if stream == CommandOutputStream::Stdout {
                            callback_bytes.lock().unwrap().extend(bytes);
                        }
                        Ok(())
                    })
                })),
                stream_output_bytes_cap: Some(8),
                ..ExecStreamingRequest::new("printf 'abcdefghijklmnopqrst'")
            })
            .await
            .unwrap();

        assert_eq!(result.result.stdout, "abcdqrst");
        assert_eq!(result.stdout_capture.observed_bytes, 20);
        assert_eq!(result.stdout_capture.retained_bytes, 8);
        assert_eq!(result.stdout_capture.omitted_bytes, 12);
        assert_eq!(&*callback_bytes.lock().unwrap(), b"abcdefghijklmnopqrst");
        std::fs::remove_dir_all(&dir).unwrap();
    }

    #[tokio::test]
    async fn exec_command_streaming_writes_exact_stdin_and_closes_it() {
        let dir = temp_dir();
        let sandbox = LocalSandbox::new(dir.clone());
        let stdin = b"first line\n$(touch must-not-run)\nlast line".to_vec();

        let result = sandbox
            .exec_command_streaming(ExecStreamingRequest {
                timeout_ms: Some(5000),
                stdin: Some(stdin.clone()),
                ..ExecStreamingRequest::new("cat")
            })
            .await
            .unwrap();

        assert_eq!(result.result.exit_code, Some(0));
        assert_eq!(result.result.stdout.as_bytes(), stdin);
        assert!(!dir.join("must-not-run").exists());
        std::fs::remove_dir_all(&dir).unwrap();
    }

    #[tokio::test]
    async fn stdio_process_evaluates_bash_before_exec() {
        let dir = temp_dir();
        let sandbox = LocalSandbox::new(dir.clone());

        // The sandbox prepends `exec`; `exec` with only a redirection applies
        // it without replacing the shell, so the Bash-only guard runs in the
        // wrapping shell before it execs the requested process.
        let process = sandbox
            .spawn_stdio_process(
                "3>&1; [[ -d / ]] || exit 9; exec python3 -u -c 'import sys; \
                 [print(line.strip()[::-1], flush=True) for line in sys.stdin]'",
                None,
                None,
                None,
            )
            .await
            .unwrap();

        let mut stdin = process.stdin;
        let mut stdout = BufReader::new(process.stdout);
        stdin.write_all(b"abc\n").await.unwrap();
        stdin.flush().await.unwrap();

        let mut line = String::new();
        stdout.read_line(&mut line).await.unwrap();
        assert_eq!(line.trim_end(), "cba");

        process.handle.terminate().await.unwrap();
        std::fs::remove_dir_all(&dir).unwrap();
    }

    #[tokio::test]
    async fn every_command_path_uses_clean_non_login_bash() {
        let dir = temp_dir();
        let sandbox = LocalSandbox::new(dir.clone());
        let bash_env_path = dir.join("bash-env");
        std::fs::write(&bash_env_path, "printf 'startup-source-loaded\\n'\n").unwrap();
        let env_vars = HashMap::from([(
            BASH_ENV_VAR.to_string(),
            bash_env_path.to_string_lossy().into_owned(),
        )]);

        let non_streaming = sandbox
            .exec_command(LOGIN_SHELL_REPORT, 5000, None, Some(&env_vars), None)
            .await
            .unwrap();
        assert_eq!(non_streaming.stdout.trim(), "nonlogin");

        let streaming = sandbox
            .exec_command_streaming(ExecStreamingRequest {
                timeout_ms: Some(5000),
                env_vars: Some(&env_vars),
                output_callback: Some(Arc::new(|_, _| Box::pin(async { Ok(()) }))),
                ..ExecStreamingRequest::new(LOGIN_SHELL_REPORT)
            })
            .await
            .unwrap();
        assert_eq!(streaming.result.stdout.trim(), "nonlogin");

        // `spawn_stdio_process` prepends `exec`, and `exec` with only a
        // redirection applies it without replacing the shell — so the rest of
        // this script reports on the wrapping shell itself.
        let process = sandbox
            .spawn_stdio_process(
                &format!("3>&1; {LOGIN_SHELL_REPORT}; exec cat"),
                None,
                Some(&env_vars),
                None,
            )
            .await
            .unwrap();
        let mut stdout = BufReader::new(process.stdout);
        let mut line = String::new();
        stdout.read_line(&mut line).await.unwrap();
        assert_eq!(line.trim_end(), "nonlogin");
        process.handle.terminate().await.unwrap();

        std::fs::remove_dir_all(&dir).unwrap();
    }

    #[tokio::test]
    async fn initialize_fails_without_bash_instead_of_reporting_ready() {
        use std::sync::Mutex;

        use crate::SandboxEvent;

        let dir = temp_dir();
        let events: Arc<Mutex<Vec<SandboxEvent>>> = Arc::new(Mutex::new(Vec::new()));
        let events_clone = Arc::clone(&events);

        let mut sandbox =
            LocalSandbox::with_bash_executable(dir.clone(), BashExecutable::Unavailable);
        sandbox.set_event_callback(Arc::new(move |e| {
            events_clone.lock().unwrap().push(e);
        }));

        let err = sandbox
            .initialize()
            .await
            .expect_err("initialize should fail without Bash");
        assert!(
            err.display_with_causes()
                .contains("Bash on the worker PATH"),
            "missing Bash should be actionable: {}",
            err.display_with_causes()
        );

        let captured = events.lock().unwrap();
        assert!(
            matches!(&captured[0], SandboxEvent::Initializing { provider } if provider == "local")
        );
        assert!(matches!(
            &captured[1],
            SandboxEvent::InitializeFailed { provider, .. } if provider == "local"
        ));
        assert!(
            !captured
                .iter()
                .any(|event| matches!(event, SandboxEvent::Ready { .. })),
            "a sandbox without Bash must never report ready: {captured:?}"
        );
        drop(captured);

        std::fs::remove_dir_all(&dir).unwrap();
    }

    #[tokio::test]
    async fn initialize_fails_when_the_resolved_executable_is_not_bash() {
        let dir = temp_dir();
        let sandbox = LocalSandbox::with_bash_executable(
            dir.clone(),
            BashExecutable::Fixed(PathBuf::from("/usr/bin/true")),
        );

        let err = sandbox
            .initialize()
            .await
            .expect_err("a non-Bash interpreter should fail the probe");

        assert!(
            err.display_with_causes().contains("non-login Bash"),
            "non-Bash interpreter should be named as the problem: {}",
            err.display_with_causes()
        );
        std::fs::remove_dir_all(&dir).unwrap();
    }

    #[tokio::test]
    async fn start_recreates_missing_working_directory_before_bash_probe() {
        let dir = temp_dir();
        std::fs::remove_dir_all(&dir).unwrap();
        let sandbox = LocalSandbox::new(dir.clone());

        sandbox
            .start()
            .await
            .expect("resume should recreate the working directory before probing Bash");

        assert!(dir.is_dir());
        std::fs::remove_dir_all(&dir).unwrap();
    }

    #[tokio::test]
    async fn start_repeats_the_bash_probe_on_resume() {
        use std::sync::Mutex;

        let dir = temp_dir();
        let successful_events: Arc<Mutex<Vec<SandboxEvent>>> = Arc::new(Mutex::new(Vec::new()));
        let successful_events_clone = Arc::clone(&successful_events);
        let mut available = LocalSandbox::new(dir.clone());
        available.set_event_callback(Arc::new(move |event| {
            successful_events_clone.lock().unwrap().push(event);
        }));

        available
            .start()
            .await
            .expect("resume should succeed when Bash is present");

        let failed_events: Arc<Mutex<Vec<SandboxEvent>>> = Arc::new(Mutex::new(Vec::new()));
        let failed_events_clone = Arc::clone(&failed_events);
        let mut unavailable =
            LocalSandbox::with_bash_executable(dir.clone(), BashExecutable::Unavailable);
        unavailable.set_event_callback(Arc::new(move |event| {
            failed_events_clone.lock().unwrap().push(event);
        }));
        unavailable
            .start()
            .await
            .expect_err("resume should fail when Bash disappeared between runs");

        let successful_events = successful_events.lock().unwrap();
        assert!(matches!(
            &successful_events[..],
            [
                SandboxEvent::StartStarted { provider: started },
                SandboxEvent::StartCompleted {
                    provider: completed,
                    ..
                }
            ] if started == "local" && completed == "local"
        ));
        let failed_events = failed_events.lock().unwrap();
        assert!(matches!(
            &failed_events[..],
            [
                SandboxEvent::StartStarted { provider: started },
                SandboxEvent::StartFailed {
                    provider: failed,
                    ..
                }
            ] if started == "local" && failed == "local"
        ));

        std::fs::remove_dir_all(&dir).unwrap();
    }

    #[tokio::test]
    async fn exec_command_exit_code() {
        let dir = temp_dir();
        let env = LocalSandbox::new(dir.clone());
        let result = env
            .exec_command("exit 42", 5000, None, None, None)
            .await
            .unwrap();

        assert_eq!(result.exit_code, Some(42));
        assert_eq!(result.termination, CommandTermination::Exited);
        std::fs::remove_dir_all(&dir).unwrap();
    }

    #[tokio::test]
    async fn exec_command_timeout() {
        let dir = temp_dir();
        let env = LocalSandbox::new(dir.clone());
        let result = env
            .exec_command("sleep 10", 200, None, None, None)
            .await
            .unwrap();

        assert_eq!(result.termination, CommandTermination::TimedOut);
        assert_eq!(result.exit_code, None);
        std::fs::remove_dir_all(&dir).unwrap();
    }

    #[tokio::test]
    async fn exec_command_cancelled() {
        let dir = temp_dir();
        let env = LocalSandbox::new(dir.clone());
        let token = CancellationToken::new();
        token.cancel();
        let result = env
            .exec_command("sleep 10", 5000, None, None, Some(token))
            .await
            .unwrap();

        assert_eq!(result.termination, CommandTermination::Cancelled);
        assert_eq!(result.exit_code, None);
        std::fs::remove_dir_all(&dir).unwrap();
    }

    #[tokio::test]
    async fn exec_command_stderr() {
        let dir = temp_dir();
        let env = LocalSandbox::new(dir.clone());
        let result = env
            .exec_command("echo err >&2", 5000, None, None, None)
            .await
            .unwrap();

        assert_eq!(result.stderr.trim(), "err");
        std::fs::remove_dir_all(&dir).unwrap();
    }

    #[tokio::test]
    async fn exec_command_filters_sensitive_explicit_env_vars() {
        let dir = temp_dir();
        let env = LocalSandbox::new(dir.clone());
        let extra = HashMap::from([
            ("FABRO_WORKER_TOKEN".to_string(), "leaked".to_string()),
            ("MY_VAR".to_string(), "ok".to_string()),
        ]);
        let result = env
            .exec_command("env", 5000, None, Some(&extra), None)
            .await
            .unwrap();

        assert!(!result.stdout.contains("FABRO_WORKER_TOKEN=leaked"));
        assert!(result.stdout.contains("MY_VAR=ok"));
        std::fs::remove_dir_all(&dir).unwrap();
    }

    #[test]
    fn env_var_filtering() {
        assert!(LocalSandbox::should_filter_env_var("OPENAI_API_KEY"));
        assert!(LocalSandbox::should_filter_env_var("ANTHROPIC_API_KEY"));
        assert!(LocalSandbox::should_filter_env_var("DB_PASSWORD"));
        assert!(LocalSandbox::should_filter_env_var("AWS_SECRET"));
        assert!(LocalSandbox::should_filter_env_var("AUTH_TOKEN"));
        assert!(LocalSandbox::should_filter_env_var("MY_CREDENTIAL"));
        assert!(LocalSandbox::should_filter_env_var("FABRO_WORKER_TOKEN"));
        assert!(LocalSandbox::should_filter_env_var("SESSION_SECRET"));
        // Case insensitive
        assert!(LocalSandbox::should_filter_env_var("my_api_key"));
        assert!(LocalSandbox::should_filter_env_var("Some_Secret"));
        // Should not filter
        assert!(!LocalSandbox::should_filter_env_var("PATH"));
        assert!(!LocalSandbox::should_filter_env_var("HOME"));
        assert!(!LocalSandbox::should_filter_env_var("EDITOR"));
        assert!(!LocalSandbox::should_filter_env_var("SECRET_PATH"));
    }

    #[test]
    fn platform_is_known() {
        let env = LocalSandbox::new(PathBuf::from("/tmp"));
        let platform = env.platform();
        assert!(
            platform == "darwin" || platform == "linux" || platform == "windows",
            "Unknown platform: {platform}"
        );
    }

    #[test]
    fn os_version_contains_platform() {
        let env = LocalSandbox::new(PathBuf::from("/tmp"));
        let version = env.os_version();
        assert!(
            version.contains(env.platform()),
            "OS version should contain platform: {version}"
        );
    }

    #[test]
    fn working_directory_accessor() {
        let env = LocalSandbox::new(PathBuf::from("/tmp/test_dir"));
        assert_eq!(env.working_directory(), "/tmp/test_dir");
    }

    #[tokio::test]
    async fn initialize_creates_directory() {
        let dir = std::env::temp_dir().join(format!("init_test_{}", uuid::Uuid::new_v4()));
        let env = LocalSandbox::new(dir.clone());
        env.initialize().await.unwrap();
        assert!(dir.exists());
        std::fs::remove_dir_all(&dir).unwrap();
    }

    #[tokio::test]
    async fn initialize_emits_events() {
        use std::sync::{Arc, Mutex};

        use crate::SandboxEvent;

        let dir = std::env::temp_dir().join(format!("init_event_test_{}", uuid::Uuid::new_v4()));
        let events: Arc<Mutex<Vec<SandboxEvent>>> = Arc::new(Mutex::new(Vec::new()));
        let events_clone = Arc::clone(&events);

        let mut env = LocalSandbox::new(dir.clone());
        env.set_event_callback(Arc::new(move |e| {
            events_clone.lock().unwrap().push(e);
        }));

        env.initialize().await.unwrap();

        let captured = events.lock().unwrap();
        assert_eq!(captured.len(), 2);
        assert!(
            matches!(&captured[0], SandboxEvent::Initializing { provider } if provider == "local")
        );
        assert!(
            matches!(&captured[1], SandboxEvent::Ready { provider, .. } if provider == "local")
        );

        std::fs::remove_dir_all(&dir).unwrap();
    }

    #[tokio::test]
    async fn cleanup_emits_events() {
        use std::sync::{Arc, Mutex};

        use crate::SandboxEvent;

        let dir = temp_dir();
        let events: Arc<Mutex<Vec<SandboxEvent>>> = Arc::new(Mutex::new(Vec::new()));
        let events_clone = Arc::clone(&events);

        let mut env = LocalSandbox::new(dir.clone());
        env.set_event_callback(Arc::new(move |e| {
            events_clone.lock().unwrap().push(e);
        }));

        env.cleanup().await.unwrap();

        let captured = events.lock().unwrap();
        assert_eq!(captured.len(), 2);
        assert!(
            matches!(&captured[0], SandboxEvent::CleanupStarted { provider } if provider == "local")
        );
        assert!(
            matches!(&captured[1], SandboxEvent::CleanupCompleted { provider, .. } if provider == "local")
        );

        std::fs::remove_dir_all(&dir).unwrap();
    }

    #[tokio::test]
    async fn stop_emits_events() {
        use std::sync::{Arc, Mutex};

        use crate::SandboxEvent;

        let dir = temp_dir();
        let events: Arc<Mutex<Vec<SandboxEvent>>> = Arc::new(Mutex::new(Vec::new()));
        let events_clone = Arc::clone(&events);

        let mut env = LocalSandbox::new(dir.clone());
        env.set_event_callback(Arc::new(move |e| {
            events_clone.lock().unwrap().push(e);
        }));

        env.stop().await.unwrap();

        let captured = events.lock().unwrap();
        assert_eq!(captured.len(), 2);
        assert!(
            matches!(&captured[0], SandboxEvent::StopStarted { provider } if provider == "local")
        );
        assert!(
            matches!(&captured[1], SandboxEvent::StopCompleted { provider, .. } if provider == "local")
        );

        std::fs::remove_dir_all(&dir).unwrap();
    }

    #[tokio::test]
    async fn grep_finds_matches() {
        let dir = temp_dir();
        std::fs::write(
            dir.join("test.rs"),
            "fn main() {\n    println!(\"hello\");\n}\n",
        )
        .unwrap();

        let env = LocalSandbox::new(dir.clone());
        let results = env
            .grep("println", "test.rs", &GrepOptions::default())
            .await
            .unwrap();

        assert_eq!(results.len(), 1);
        assert!(results[0].contains("println"));
        std::fs::remove_dir_all(&dir).unwrap();
    }

    #[tokio::test]
    async fn grep_case_insensitive() {
        let dir = temp_dir();
        std::fs::write(dir.join("test.txt"), "Hello\nhello\nHELLO\n").unwrap();

        let env = LocalSandbox::new(dir.clone());
        let results = env
            .grep("hello", "test.txt", &GrepOptions {
                case_insensitive: true,
                ..Default::default()
            })
            .await
            .unwrap();

        assert_eq!(results.len(), 3);
        std::fs::remove_dir_all(&dir).unwrap();
    }

    #[tokio::test]
    async fn grep_max_results() {
        let dir = temp_dir();
        std::fs::write(dir.join("test.txt"), "match1\nmatch2\nmatch3\nmatch4\n").unwrap();

        let env = LocalSandbox::new(dir.clone());
        let results = env
            .grep("match", "test.txt", &GrepOptions {
                max_results: Some(2),
                ..Default::default()
            })
            .await
            .unwrap();

        assert_eq!(results.len(), 2);
        std::fs::remove_dir_all(&dir).unwrap();
    }

    #[tokio::test]
    async fn glob_finds_files() {
        let dir = temp_dir();
        std::fs::write(dir.join("a.rs"), "").unwrap();
        std::fs::write(dir.join("b.rs"), "").unwrap();
        std::fs::write(dir.join("c.txt"), "").unwrap();

        let env = LocalSandbox::new(dir.clone());
        let results = env.glob("*.rs", None).await.unwrap();

        assert_eq!(results.len(), 2);
        std::fs::remove_dir_all(&dir).unwrap();
    }

    #[tokio::test]
    async fn glob_resolves_relative_search_path_against_working_directory() {
        let dir = temp_dir();
        std::fs::create_dir_all(dir.join("src")).unwrap();
        std::fs::write(dir.join("src/lib.rs"), "").unwrap();

        let env = LocalSandbox::new(dir.clone());
        let results = env.glob("*.rs", Some("src")).await.unwrap();

        assert_eq!(results, vec![
            dir.join("src/lib.rs").to_string_lossy().into_owned()
        ]);
        std::fs::remove_dir_all(&dir).unwrap();
    }

    #[tokio::test]
    async fn glob_recursive_pattern_finds_files_at_any_depth() {
        let dir = temp_dir();
        std::fs::create_dir_all(dir.join("src/nested")).unwrap();
        std::fs::write(dir.join("a.rs"), "").unwrap();
        std::fs::write(dir.join("src/lib.rs"), "").unwrap();
        std::fs::write(dir.join("src/nested/main.rs"), "").unwrap();
        std::fs::write(dir.join("src/nested/readme.md"), "").unwrap();

        let env = LocalSandbox::new(dir.clone());
        let results = env.glob("**/*.rs", None).await.unwrap();

        assert_eq!(results, vec![
            dir.join("a.rs").to_string_lossy().into_owned(),
            dir.join("src/lib.rs").to_string_lossy().into_owned(),
            dir.join("src/nested/main.rs")
                .to_string_lossy()
                .into_owned(),
        ]);
        std::fs::remove_dir_all(&dir).unwrap();
    }

    #[tokio::test]
    async fn glob_finds_skill_files_one_level_below_search_dir() {
        let dir = temp_dir();
        let skills = dir.join(".fabro/skills");
        std::fs::create_dir_all(skills.join("patch")).unwrap();
        std::fs::create_dir_all(skills.join("nested/deeper")).unwrap();
        std::fs::write(skills.join("SKILL.md"), "").unwrap();
        std::fs::write(skills.join("patch/SKILL.md"), "").unwrap();
        std::fs::write(skills.join("nested/deeper/SKILL.md"), "").unwrap();

        let env = LocalSandbox::new(dir.clone());
        let skills_path = skills.to_string_lossy().into_owned();
        let results = env.glob("*/SKILL.md", Some(&skills_path)).await.unwrap();

        assert_eq!(results, vec![
            skills.join("patch/SKILL.md").to_string_lossy().into_owned()
        ]);
        std::fs::remove_dir_all(&dir).unwrap();
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn glob_does_not_recurse_through_symlinked_directories() {
        let dir = temp_dir();
        let target = dir.join("target");
        std::fs::create_dir_all(&target).unwrap();
        std::fs::write(target.join("lib.rs"), "").unwrap();
        std::os::unix::fs::symlink(&target, dir.join("linked")).unwrap();

        let env = LocalSandbox::new(dir.clone());
        let results = env.glob("linked/**/*.rs", None).await.unwrap();

        assert!(results.is_empty());
        std::fs::remove_dir_all(&dir).unwrap();
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn glob_follows_the_declared_symlinked_search_root_only() {
        let parent = temp_dir();
        let workspace = parent.join("workspace");
        let outside = parent.join("outside");
        let search_root = parent.join("workspace-link");
        std::fs::create_dir_all(&workspace).unwrap();
        std::fs::create_dir_all(&outside).unwrap();
        std::fs::write(workspace.join("README.md"), "").unwrap();
        std::fs::write(outside.join("outside.md"), "").unwrap();
        std::os::unix::fs::symlink(&outside, workspace.join("linked")).unwrap();
        std::os::unix::fs::symlink(&workspace, &search_root).unwrap();

        let env = LocalSandbox::new(search_root.clone());
        let results = env.glob("**/*.md", None).await.unwrap();

        assert_eq!(results, vec![
            search_root.join("README.md").to_string_lossy().into_owned()
        ]);
        std::fs::remove_dir_all(&parent).unwrap();
    }

    #[tokio::test]
    async fn walk_files_returns_relative_paths_and_prunes_directories() {
        let dir = temp_dir();
        std::fs::create_dir_all(dir.join(".ai/reports")).unwrap();
        std::fs::create_dir_all(dir.join(".ai/target")).unwrap();
        std::fs::write(dir.join(".ai/reports/result.md"), "report").unwrap();
        std::fs::write(dir.join(".ai/reports/empty.md"), "").unwrap();
        std::fs::write(dir.join(".ai/target/ignored.md"), "ignored").unwrap();

        let sandbox = LocalSandbox::new(dir.clone());
        let files = sandbox
            .walk_files(sandbox.working_directory(), ".ai", &WalkOptions {
                excluded_directory_names: vec!["target".to_string()],
            })
            .await
            .unwrap();

        let mut file_metadata = files
            .iter()
            .map(|file| (file.relative_path.as_str(), file.size))
            .collect::<Vec<_>>();
        file_metadata.sort_unstable();
        assert_eq!(file_metadata, vec![
            (".ai/reports/empty.md", 0),
            (".ai/reports/result.md", 6),
        ]);
        let root = dir.to_string_lossy();
        assert!(
            files
                .iter()
                .all(|file| file.path.starts_with(root.as_ref()))
        );
        std::fs::remove_dir_all(&dir).unwrap();
    }

    #[tokio::test]
    async fn local_sandbox_download_file_to_local() {
        let dir = temp_dir();
        std::fs::write(dir.join("source.txt"), "hello download").unwrap();

        let env = LocalSandbox::new(dir.clone());
        let dest = dir.join("output/downloaded.txt");
        env.download_file_to_local("source.txt", &dest)
            .await
            .unwrap();

        assert_eq!(std::fs::read_to_string(&dest).unwrap(), "hello download");
        std::fs::remove_dir_all(&dir).unwrap();
    }

    #[tokio::test]
    async fn local_sandbox_download_file_to_local_creates_parent_dirs() {
        let dir = temp_dir();
        std::fs::write(dir.join("data.bin"), "binary-ish").unwrap();

        let env = LocalSandbox::new(dir.clone());
        let dest = dir.join("deep/nested/dir/data.bin");
        env.download_file_to_local("data.bin", &dest).await.unwrap();

        assert_eq!(std::fs::read_to_string(&dest).unwrap(), "binary-ish");
        std::fs::remove_dir_all(&dir).unwrap();
    }

    #[tokio::test]
    async fn local_sandbox_download_file_to_local_binary() {
        let dir = temp_dir();
        let binary_data: Vec<u8> = (0u8..=255).collect();
        std::fs::write(dir.join("binary.bin"), &binary_data).unwrap();

        let env = LocalSandbox::new(dir.clone());
        let dest = dir.join("out/binary.bin");
        env.download_file_to_local("binary.bin", &dest)
            .await
            .unwrap();

        assert_eq!(std::fs::read(&dest).unwrap(), binary_data);
        std::fs::remove_dir_all(&dir).unwrap();
    }
}
