agent_workspace_adapter_local/

lib.rs

//! Explicit local/test workspace adapter.
//!
//! This crate provides a concrete implementation of the [`Workspace`] trait
//! for local filesystem access. It is the counterpart of `agent-workspace-server`
//! for single-process deployments and testing.
//!
//! # Design decisions
//!
//! ## Local-first, portable
//!
//! The local workspace uses only `tokio::fs` and `tokio::process::Command`,
//! requiring no external dependencies beyond the host filesystem. Remote
//! providers (Daytona, Moulin, E2B) are in the `agents-runtime` crate and
//! may be migrated here in a follow-up.
//!
//! ## Path safety
//!
//! All paths are resolved relative to the workspace root. Absolute paths are
//! accepted but must be within the root or the operation fails.

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

#[cfg(not(unix))]
use agent_workspace_contract::WORKSPACE_MAX_FILE_BYTES;
use agent_workspace_contract::{
    CmdOutput, DirEntry, FileVisibilityOptions, GrepMatch, PortUrl, Workspace,
    WORKSPACE_MAX_COMMAND_BYTES, WORKSPACE_MAX_COMMAND_OUTPUT_BYTES, WORKSPACE_MAX_RESULTS,
    WORKSPACE_MAX_SEARCH_OUTPUT_BYTES,
};
use anyhow::{Context, Result};
use async_trait::async_trait;
#[cfg(not(unix))]
use sha2::{Digest as _, Sha256};
#[cfg(not(unix))]
use tokio::io::AsyncWriteExt;
use tokio::io::{AsyncRead, AsyncReadExt};

mod executor;
mod local_environment;
pub use local_environment::*;
mod path_resolver;
mod provider;
mod search;
#[cfg(unix)]
mod secure_fs;
mod snapshot_object_store;
mod snapshot_provider;
pub use executor::*;
use path_resolver::{normalize_path, WorkspacePathResolver};
pub use provider::*;
#[cfg(unix)]
use secure_fs::SecureRoot;
pub use snapshot_object_store::*;
pub use snapshot_provider::*;

const COMMAND_TIMEOUT: Duration = Duration::from_secs(60);

/// A local filesystem workspace implementation.
///
/// All operations are performed against the host filesystem under the
/// configured `root` directory.
#[derive(Debug, Clone)]
pub struct LocalWorkspace {
    root: PathBuf,
    allow_host_exec: bool,
    #[cfg(unix)]
    secure_root: SecureRoot,
}

#[cfg(test)]
mod tests;

impl LocalWorkspace {
    pub async fn write_upload_chunk(&self, path: &str, offset: u64, bytes: &[u8]) -> Result<u64> {
        #[cfg(unix)]
        {
            let secure = self.secure_root.clone();
            let path = path.to_string();
            let bytes = bytes.to_vec();
            tokio::task::spawn_blocking(move || secure.write_chunk(&path, offset, &bytes))
                .await
                .context("joining descriptor-relative upload write")?
        }
        #[cfg(not(unix))]
        {
            let mut current = if self.exists(path).await? {
                self.read_file_bytes(path).await?
            } else {
                Vec::new()
            };
            anyhow::ensure!(current.len() as u64 == offset, "upload offset conflict");
            current.extend_from_slice(bytes);
            self.write_file_bytes(path, &current).await?;
            Ok(current.len() as u64)
        }
    }
    pub fn new(root: PathBuf) -> Self {
        Self::new_with_host_exec(root, false)
    }

    /// Explicitly opt into host command execution for trusted local/test use.
    pub fn new_with_host_exec(root: PathBuf, allow_host_exec: bool) -> Self {
        let root = if root.is_absolute() {
            root
        } else {
            std::env::current_dir()
                .map(|current| current.join(&root))
                .unwrap_or(root)
        };
        Self {
            root: normalize_path(&root),
            allow_host_exec,
            #[cfg(unix)]
            secure_root: SecureRoot::new(normalize_path(&root)),
        }
    }

    pub fn capabilities(&self) -> agent_workspace_contract::WorkspaceCapabilities {
        agent_workspace_contract::WorkspaceCapabilities {
            commands: self.allow_host_exec,
            previews: true,
            atomic_writes: true,
            compare_and_swap: true,
            ..Default::default()
        }
    }

    /// Resolve a user-supplied path lexically against the workspace root.
    fn resolve(&self, path: &str) -> Result<PathBuf> {
        WorkspacePathResolver::new(&self.root).lexical(path)
    }

    #[cfg(not(unix))]
    async fn canonical_root(&self) -> Result<PathBuf> {
        WorkspacePathResolver::new(&self.root)
            .canonical_root()
            .await
    }

    async fn resolve_existing(&self, path: &str) -> Result<PathBuf> {
        WorkspacePathResolver::new(&self.root).existing(path).await
    }

    #[cfg(not(unix))]
    async fn resolve_for_write(&self, path: &str) -> Result<PathBuf> {
        WorkspacePathResolver::new(&self.root).for_write(path).await
    }

    #[cfg(not(unix))]
    async fn atomic_write_unlocked(&self, path: &str, content: &[u8]) -> Result<()> {
        anyhow::ensure!(
            content.len() <= WORKSPACE_MAX_FILE_BYTES,
            "workspace file is too large"
        );
        let target = self.resolve_for_write(path).await?;
        let parent = target.parent().context("workspace file has no parent")?;
        tokio::fs::create_dir_all(parent).await?;
        // Re-resolve after creating parents. This closes the common symlink
        // substitution window before the same-directory atomic rename.
        let target = self.resolve_for_write(path).await?;
        let temp = parent.join(format!(
            ".agent-write-{}.tmp",
            uuid::Uuid::new_v4().simple()
        ));
        let result = async {
            let mut file = tokio::fs::OpenOptions::new()
                .create_new(true)
                .write(true)
                .open(&temp)
                .await?;
            file.write_all(content).await?;
            file.sync_all().await?;
            drop(file);
            tokio::fs::rename(&temp, &target).await?;
            Ok::<_, anyhow::Error>(())
        }
        .await;
        if result.is_err() {
            let _ = tokio::fs::remove_file(&temp).await;
        }
        result.with_context(|| format!("failed to atomically write {}", target.display()))
    }

    #[cfg(not(unix))]
    async fn revision_for_existing(&self, path: &Path) -> Result<String> {
        let mut file = tokio::fs::File::open(path).await?;
        let mut hash = Sha256::new();
        let mut total = 0_usize;
        let mut buffer = [0_u8; 16 * 1024];
        loop {
            let read = file.read(&mut buffer).await?;
            if read == 0 {
                break;
            }
            total = total.saturating_add(read);
            anyhow::ensure!(
                total <= WORKSPACE_MAX_FILE_BYTES,
                "workspace file exceeds {WORKSPACE_MAX_FILE_BYTES} bytes"
            );
            hash.update(&buffer[..read]);
        }
        Ok(format!("sha256:{:x}", hash.finalize()))
    }

    async fn list_dir_inner(
        &self,
        path: &str,
        options: Option<FileVisibilityOptions>,
    ) -> Result<Vec<DirEntry>> {
        #[cfg(unix)]
        {
            let secure = self.secure_root.clone();
            let path = path.to_string();
            tokio::task::spawn_blocking(move || secure.list_dir(&path, options))
                .await
                .context("joining descriptor-relative directory listing")?
        }
        #[cfg(not(unix))]
        {
            let target = self.resolve_existing(path).await?;
            let mut entries = Vec::new();
            let mut read_dir = tokio::fs::read_dir(&target)
                .await
                .with_context(|| format!("failed to list {}", target.display()))?;
            while let Some(entry) = read_dir.next_entry().await? {
                anyhow::ensure!(
                    entries.len() < WORKSPACE_MAX_RESULTS,
                    "directory contains too many entries"
                );
                let name = entry.file_name().to_string_lossy().to_string();
                if options
                    .map(|options| !search::is_visible_name(&name, options))
                    .unwrap_or(false)
                {
                    continue;
                }
                let metadata = tokio::fs::symlink_metadata(entry.path()).await?;
                entries.push(DirEntry {
                    name,
                    is_dir: metadata.is_dir(),
                    size: if metadata.is_file() {
                        Some(metadata.len())
                    } else {
                        None
                    },
                });
            }
            entries.sort_by(|a, b| a.name.cmp(&b.name));
            Ok(entries)
        }
    }

    async fn walk_tree_inner(
        &self,
        path: &str,
        max_depth: usize,
        options: Option<FileVisibilityOptions>,
    ) -> Result<Vec<String>> {
        #[cfg(unix)]
        {
            let secure = self.secure_root.clone();
            let path = path.to_string();
            tokio::task::spawn_blocking(move || secure.walk(&path, max_depth, options))
                .await
                .context("joining descriptor-relative tree walk")?
        }
        #[cfg(not(unix))]
        {
            let target = self.resolve_existing(path).await?;
            let mut results = Vec::new();
            let mut output_bytes = 0;
            let root = self.canonical_root().await?;
            self.walk_dir(
                &target,
                max_depth.min(64),
                &mut results,
                &mut output_bytes,
                &root,
                options,
            )?;
            Ok(results)
        }
    }

    #[cfg(not(unix))]
    fn walk_dir(
        &self,
        dir: &Path,
        remaining_depth: usize,
        results: &mut Vec<String>,
        output_bytes: &mut usize,
        root: &Path,
        options: Option<FileVisibilityOptions>,
    ) -> Result<()> {
        if remaining_depth == 0 {
            return Ok(());
        }
        let read_dir =
            std::fs::read_dir(dir).with_context(|| format!("failed to list {}", dir.display()))?;
        for entry in read_dir {
            let entry = entry?;
            let name = entry.file_name().to_string_lossy().to_string();
            if options
                .map(|opt| !search::is_visible_name(&name, opt))
                .unwrap_or(false)
            {
                continue;
            }
            let path = entry.path();
            anyhow::ensure!(
                results.len() < WORKSPACE_MAX_RESULTS,
                "workspace tree result limit exceeded"
            );
            let relative = relative_path(root, &path)?;
            *output_bytes = output_bytes.saturating_add(relative.len());
            anyhow::ensure!(
                *output_bytes <= WORKSPACE_MAX_SEARCH_OUTPUT_BYTES,
                "workspace tree output limit exceeded"
            );
            results.push(relative);
            if entry.file_type()?.is_dir() {
                self.walk_dir(
                    &path,
                    remaining_depth - 1,
                    results,
                    output_bytes,
                    root,
                    options,
                )?;
            }
        }
        Ok(())
    }
}

#[async_trait]
impl agent_workspace_contract::WorkspaceUploadWriter for LocalWorkspace {
    async fn write_upload_chunk(&self, path: &str, offset: u64, bytes: &[u8]) -> Result<u64> {
        LocalWorkspace::write_upload_chunk(self, path, offset, bytes).await
    }
}

#[async_trait]
impl Workspace for LocalWorkspace {
    fn description(&self) -> String {
        format!("local:{}", self.root.display())
    }

    fn root(&self) -> PathBuf {
        self.root.clone()
    }

    fn capabilities(&self) -> agent_workspace_contract::WorkspaceCapabilities {
        LocalWorkspace::capabilities(self)
    }

    async fn read_file(&self, path: &str) -> Result<String> {
        let bytes = self.read_file_bytes(path).await?;
        String::from_utf8(bytes).context("workspace file is not valid UTF-8")
    }

    async fn read_file_bytes(&self, path: &str) -> Result<Vec<u8>> {
        #[cfg(unix)]
        {
            let secure = self.secure_root.clone();
            let path = path.to_string();
            return tokio::task::spawn_blocking(move || secure.read(&path))
                .await
                .context("joining descriptor-relative read")?;
        }
        #[cfg(not(unix))]
        {
            let target = self.resolve_existing(path).await?;
            let file = tokio::fs::File::open(&target)
                .await
                .with_context(|| format!("failed to read {}", target.display()))?;
            let (bytes, truncated) = read_bounded(file, WORKSPACE_MAX_FILE_BYTES).await?;
            anyhow::ensure!(
                !truncated,
                "workspace file exceeds {WORKSPACE_MAX_FILE_BYTES} bytes"
            );
            Ok(bytes)
        }
    }

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

    async fn write_file_bytes(&self, path: &str, content: &[u8]) -> Result<()> {
        self.write_file_if_match(path, content, None)
            .await
            .map(|_| ())
    }

    async fn file_revision(&self, path: &str) -> Result<Option<String>> {
        #[cfg(unix)]
        {
            let secure = self.secure_root.clone();
            let path = path.to_string();
            return tokio::task::spawn_blocking(move || secure.revision(&path))
                .await
                .context("joining descriptor-relative revision")?;
        }
        #[cfg(not(unix))]
        {
            match tokio::fs::symlink_metadata(self.resolve(path)?).await {
                Ok(_) => {
                    let target = self.resolve_existing(path).await?;
                    self.revision_for_existing(&target).await.map(Some)
                }
                Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(None),
                Err(error) => Err(error.into()),
            }
        }
    }

    async fn write_file_if_match(
        &self,
        path: &str,
        content: &[u8],
        if_match: Option<&str>,
    ) -> Result<String> {
        #[cfg(unix)]
        {
            let secure = self.secure_root.clone();
            let path = path.to_string();
            let content = content.to_vec();
            let expected = if_match.map(str::to_string);
            return tokio::task::spawn_blocking(move || {
                secure.write_if_match(&path, &content, expected.as_deref())
            })
            .await
            .context("joining descriptor-relative write")?;
        }
        #[cfg(not(unix))]
        {
            if let Some(expected) = if_match {
                let actual = match tokio::fs::symlink_metadata(self.resolve(path)?).await {
                    Ok(_) => {
                        let target = self.resolve_existing(path).await?;
                        Some(self.revision_for_existing(&target).await?)
                    }
                    Err(error) if error.kind() == std::io::ErrorKind::NotFound => None,
                    Err(error) => return Err(error.into()),
                };
                if !agent_workspace_contract::revision_matches(expected, actual.as_deref()) {
                    return Err(agent_workspace_contract::WorkspaceVersionConflict {
                        expected: expected.to_string(),
                        actual,
                    }
                    .into());
                }
            }
            self.atomic_write_unlocked(path, content).await?;
            Ok(agent_workspace_contract::content_revision(content))
        }
    }

    async fn create_dir_all(&self, path: &str) -> Result<()> {
        #[cfg(unix)]
        {
            let secure = self.secure_root.clone();
            let path = path.to_string();
            return tokio::task::spawn_blocking(move || secure.create_dir_all(&path))
                .await
                .context("joining descriptor-relative mkdir")?;
        }
        #[cfg(not(unix))]
        {
            let target = self.resolve_for_write(path).await?;
            tokio::fs::create_dir_all(&target)
                .await
                .with_context(|| format!("failed to create directory {}", target.display()))
        }
    }

    async fn remove_file(&self, path: &str) -> Result<()> {
        #[cfg(unix)]
        {
            let secure = self.secure_root.clone();
            let path = path.to_string();
            return tokio::task::spawn_blocking(move || secure.remove(&path))
                .await
                .context("joining descriptor-relative remove")?;
        }
        #[cfg(not(unix))]
        {
            let target = self.resolve_existing(path).await?;
            tokio::fs::remove_file(&target)
                .await
                .with_context(|| format!("failed to remove {}", target.display()))
        }
    }

    async fn remove_file_if_match(&self, path: &str, if_match: &str) -> Result<()> {
        #[cfg(unix)]
        {
            let secure = self.secure_root.clone();
            let path = path.to_string();
            let expected = if_match.to_string();
            return tokio::task::spawn_blocking(move || {
                secure.remove_if_match(&path, Some(&expected))
            })
            .await
            .context("joining descriptor-relative conditional remove")?;
        }
        #[cfg(not(unix))]
        {
            let actual = self.file_revision(path).await?;
            if !agent_workspace_contract::revision_matches(if_match, actual.as_deref()) {
                return Err(agent_workspace_contract::WorkspaceVersionConflict {
                    expected: if_match.to_string(),
                    actual,
                }
                .into());
            }
            self.remove_file(path).await
        }
    }

    async fn exists(&self, path: &str) -> Result<bool> {
        let target = self.resolve(path)?;
        match tokio::fs::symlink_metadata(&target).await {
            Ok(_) => {
                self.resolve_existing(path).await?;
                Ok(true)
            }
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(false),
            Err(error) => Err(error.into()),
        }
    }

    async fn is_dir(&self, path: &str) -> Result<bool> {
        if !self.exists(path).await? {
            return Ok(false);
        }
        Ok(tokio::fs::metadata(self.resolve_existing(path).await?)
            .await?
            .is_dir())
    }

    async fn list_dir(&self, path: &str) -> Result<Vec<DirEntry>> {
        self.list_dir_inner(path, None).await
    }

    async fn list_dir_with_options(
        &self,
        path: &str,
        options: FileVisibilityOptions,
    ) -> Result<Vec<DirEntry>> {
        self.list_dir_inner(path, Some(options)).await
    }

    async fn walk_tree(&self, path: &str, max_depth: usize) -> Result<Vec<String>> {
        self.walk_tree_inner(path, max_depth, None).await
    }

    async fn walk_tree_with_options(
        &self,
        path: &str,
        max_depth: usize,
        options: FileVisibilityOptions,
    ) -> Result<Vec<String>> {
        self.walk_tree_inner(path, max_depth, Some(options)).await
    }

    async fn find_files(&self, pattern: &str, path: &str) -> Result<Vec<String>> {
        #[cfg(unix)]
        {
            let secure = self.secure_root.clone();
            let path = path.to_string();
            let pattern = glob::Pattern::new(pattern)?;
            return tokio::task::spawn_blocking(move || {
                let mut output_bytes = 0_usize;
                let mut result = Vec::new();
                for candidate in secure.walk(&path, 64, None)? {
                    let name = Path::new(&candidate)
                        .file_name()
                        .and_then(|value| value.to_str())
                        .unwrap_or("");
                    if pattern.matches(name) && secure.is_regular_file(&candidate)? {
                        output_bytes = output_bytes.saturating_add(candidate.len());
                        anyhow::ensure!(
                            output_bytes <= WORKSPACE_MAX_SEARCH_OUTPUT_BYTES,
                            "find output limit exceeded"
                        );
                        result.push(candidate);
                    }
                }
                Ok(result)
            })
            .await
            .context("joining descriptor-relative file search")?;
        }
        #[cfg(not(unix))]
        {
            let target = self.resolve_existing(path).await?;
            let pattern = glob::Pattern::new(pattern)?;
            let mut results = Vec::new();
            let mut output_bytes = 0;
            let root = self.canonical_root().await?;
            self.find_files_inner(&target, &pattern, &mut results, &mut output_bytes, &root)?;
            Ok(results)
        }
    }

    async fn grep(
        &self,
        pattern: &str,
        path: &str,
        include: Option<&str>,
    ) -> Result<Vec<GrepMatch>> {
        #[cfg(unix)]
        {
            let secure = self.secure_root.clone();
            let path = path.to_string();
            let regex = regex::Regex::new(pattern)?;
            let include_pattern = include.map(glob::Pattern::new).transpose()?;
            return tokio::task::spawn_blocking(move || {
                let mut result = Vec::new();
                let mut output_bytes = 0_usize;
                for candidate in secure.walk(&path, 64, None)? {
                    let name = Path::new(&candidate)
                        .file_name()
                        .and_then(|value| value.to_str())
                        .unwrap_or("");
                    if !secure.is_regular_file(&candidate)?
                        || !include_pattern
                            .as_ref()
                            .is_none_or(|value| value.matches(name))
                    {
                        continue;
                    }
                    let bytes = secure.read(&candidate)?;
                    let content = String::from_utf8_lossy(&bytes);
                    for (line_number, line) in content.lines().enumerate() {
                        if regex.is_match(line) {
                            anyhow::ensure!(
                                result.len() < WORKSPACE_MAX_RESULTS,
                                "grep result limit exceeded"
                            );
                            output_bytes = output_bytes
                                .saturating_add(candidate.len())
                                .saturating_add(line.len());
                            anyhow::ensure!(
                                output_bytes <= WORKSPACE_MAX_SEARCH_OUTPUT_BYTES,
                                "grep output limit exceeded"
                            );
                            result.push(GrepMatch {
                                file: candidate.clone(),
                                line_number: line_number + 1,
                                line: line.into(),
                            });
                        }
                    }
                }
                Ok(result)
            })
            .await
            .context("joining descriptor-relative grep")?;
        }
        #[cfg(not(unix))]
        {
            let target = self.resolve_existing(path).await?;
            let regex = regex::Regex::new(pattern)?;
            let include_pattern = include.map(glob::Pattern::new).transpose()?;
            let mut results = Vec::new();
            let mut output_bytes = 0;
            let root = self.canonical_root().await?;
            self.grep_inner(
                &target,
                &regex,
                include_pattern.as_ref(),
                &mut results,
                &mut output_bytes,
                &root,
            )?;
            Ok(results)
        }
    }

    async fn exec(&self, command: &str, cwd: Option<&str>) -> Result<CmdOutput> {
        anyhow::ensure!(
            self.allow_host_exec,
            "host command execution is disabled for this local workspace"
        );
        anyhow::ensure!(
            command.len() <= WORKSPACE_MAX_COMMAND_BYTES,
            "workspace command exceeds {} bytes",
            WORKSPACE_MAX_COMMAND_BYTES
        );
        let cwd = match cwd {
            Some(c) => self.resolve_existing(c).await?,
            None => self.resolve_existing(".").await?,
        };
        let mut child = tokio::process::Command::new("sh")
            .arg("-lc")
            .arg(command)
            .current_dir(&cwd)
            .stdout(Stdio::piped())
            .stderr(Stdio::piped())
            .kill_on_drop(true)
            .spawn()
            .with_context(|| format!("failed to execute: {command}"))?;
        let stdout = child
            .stdout
            .take()
            .context("command stdout was not captured")?;
        let stderr = child
            .stderr
            .take()
            .context("command stderr was not captured")?;
        let stdout_task = tokio::spawn(read_bounded(stdout, WORKSPACE_MAX_COMMAND_OUTPUT_BYTES));
        let stderr_task = tokio::spawn(read_bounded(stderr, WORKSPACE_MAX_COMMAND_OUTPUT_BYTES));

        let status = match tokio::time::timeout(COMMAND_TIMEOUT, child.wait()).await {
            Ok(result) => result?,
            Err(_) => {
                child.kill().await?;
                let _ = child.wait().await;
                anyhow::bail!("workspace command timed out after 60 seconds");
            }
        };
        let (stdout, stdout_truncated) = stdout_task.await.context("joining stdout reader")??;
        let (stderr, stderr_truncated) = stderr_task.await.context("joining stderr reader")??;

        Ok(CmdOutput {
            exit_code: status.code().unwrap_or(-1),
            stdout: String::from_utf8_lossy(&stdout).to_string(),
            stderr: String::from_utf8_lossy(&stderr).to_string(),
            stdout_truncated,
            stderr_truncated,
        })
    }

    async fn port_url(&self, port: u16) -> Result<PortUrl> {
        Ok(PortUrl {
            port,
            url: format!("http://localhost:{port}"),
            host: Some("localhost".to_string()),
            token: None,
            signed: false,
            expires_in_seconds: None,
        })
    }
}

impl LocalWorkspace {
    #[cfg(not(unix))]
    fn find_files_inner(
        &self,
        dir: &Path,
        pattern: &glob::Pattern,
        results: &mut Vec<String>,
        output_bytes: &mut usize,
        root: &Path,
    ) -> Result<()> {
        let read_dir =
            std::fs::read_dir(dir).with_context(|| format!("failed to list {}", dir.display()))?;
        for entry in read_dir {
            let entry = entry?;
            let path = entry.path();
            let name = entry.file_name().to_string_lossy().to_string();
            let file_type = entry.file_type()?;
            if file_type.is_dir() {
                self.find_files_inner(&path, pattern, results, output_bytes, root)?;
            } else if file_type.is_file() && pattern.matches(&name) {
                anyhow::ensure!(
                    results.len() < WORKSPACE_MAX_RESULTS,
                    "find result limit exceeded"
                );
                let relative = relative_path(root, &path)?;
                *output_bytes = output_bytes.saturating_add(relative.len());
                anyhow::ensure!(
                    *output_bytes <= WORKSPACE_MAX_SEARCH_OUTPUT_BYTES,
                    "find output limit exceeded"
                );
                results.push(relative);
            }
        }
        Ok(())
    }

    #[cfg(not(unix))]
    fn grep_inner(
        &self,
        dir: &Path,
        regex: &regex::Regex,
        include_pattern: Option<&glob::Pattern>,
        results: &mut Vec<GrepMatch>,
        output_bytes: &mut usize,
        root: &Path,
    ) -> Result<()> {
        let read_dir =
            std::fs::read_dir(dir).with_context(|| format!("failed to list {}", dir.display()))?;
        for entry in read_dir {
            let entry = entry?;
            let path = entry.path();
            let name = entry.file_name().to_string_lossy().to_string();
            let file_type = entry.file_type()?;
            if file_type.is_dir() {
                self.grep_inner(&path, regex, include_pattern, results, output_bytes, root)?;
            } else if file_type.is_file() && include_pattern.is_none_or(|p| p.matches(&name)) {
                anyhow::ensure!(
                    entry.metadata()?.len() <= WORKSPACE_MAX_FILE_BYTES as u64,
                    "grep input file is too large: {}",
                    path.display()
                );
                if let Ok(content) = std::fs::read_to_string(&path) {
                    for (line_number, line) in content.lines().enumerate() {
                        if regex.is_match(line) {
                            anyhow::ensure!(
                                results.len() < WORKSPACE_MAX_RESULTS,
                                "grep result limit exceeded"
                            );
                            let file = relative_path(root, &path)?;
                            *output_bytes = output_bytes
                                .saturating_add(file.len())
                                .saturating_add(line.len());
                            anyhow::ensure!(
                                *output_bytes <= WORKSPACE_MAX_SEARCH_OUTPUT_BYTES,
                                "grep output limit exceeded"
                            );
                            results.push(GrepMatch {
                                file,
                                line_number: line_number + 1,
                                line: line.to_string(),
                            });
                        }
                    }
                }
            }
        }
        Ok(())
    }
}

#[cfg(not(unix))]
fn relative_path(root: &Path, path: &Path) -> Result<String> {
    Ok(path
        .strip_prefix(root)
        .context("workspace traversal escaped the root")?
        .to_string_lossy()
        .to_string())
}

async fn read_bounded<Reader>(mut reader: Reader, limit: usize) -> std::io::Result<(Vec<u8>, bool)>
where
    Reader: AsyncRead + Unpin,
{
    let mut bytes = Vec::with_capacity(limit.min(64 * 1024));
    let mut buffer = [0u8; 16 * 1024];
    let mut truncated = false;
    loop {
        let read = reader.read(&mut buffer).await?;
        if read == 0 {
            break;
        }
        let remaining = limit.saturating_sub(bytes.len());
        let retained = read.min(remaining);
        bytes.extend_from_slice(&buffer[..retained]);
        truncated |= retained < read;
    }
    Ok((bytes, truncated))
}

/// Create a local workspace at the given root.
pub async fn create_local_workspace(root: PathBuf) -> Result<Box<dyn Workspace>> {
    let ws = LocalWorkspace::new(root);
    ws.init().await?;
    Ok(Box::new(ws))
}