Require filesystem backends to implement directory walks (#39749)

## What changed

- Make `ExecutorFileSystem::walk` a required backend operation instead of
  providing a fallback built from directory reads and metadata requests.
- Implement bounded local walks on a blocking task with cancellation, symlink
  cycle detection, deterministic ordering, error collection, and response-size
  limits.
- Have remote filesystems use the server's walk operation directly.

## Testing

- Cover local and remote handling of invalid roots and limits, directory
  symlinks, non-UTF-8 names, cancellation, sandbox contexts, and response
  budgets.

GitOrigin-RevId: 7499bf05080c3f9965a5eb7ffd593de604d62c2a

codex-rs/exec-server/src/local_file_system.rs

@@ -1,5 +1,12 @@
+use codex_file_system::MAX_WALK_DEPTH;
+use codex_file_system::MAX_WALK_DIRECTORIES;
+use codex_file_system::MAX_WALK_ENTRIES;
+use codex_file_system::MAX_WALK_RESPONSE_BYTES;
+use codex_file_system::WALK_RESPONSE_ITEM_OVERHEAD_BYTES;
 use codex_utils_absolute_path::AbsolutePathBuf;
 use codex_utils_path_uri::PathUri;
+use std::collections::HashSet;
+use std::collections::VecDeque;
 use std::path::Path;
 use std::path::PathBuf;
 use std::sync::Arc;
@@ -9,6 +16,7 @@ use std::time::UNIX_EPOCH;
 use tokio::io;
 use tokio::io::AsyncReadExt;
 use tokio_util::io::ReaderStream;
+use tokio_util::sync::CancellationToken;
 
 use crate::CopyOptions;
 use crate::CreateDirectoryOptions;
@@ -24,6 +32,9 @@ use crate::GetMetadataOptions;
 use crate::ReadDirectoryEntry;
 use crate::ReadFileOptions;
 use crate::RemoveOptions;
+use crate::WalkEntry;
+use crate::WalkEntryKind;
+use crate::WalkError;
 use crate::WalkOptions;
 use crate::WalkOutcome;
 use crate::WriteFileOptions;
@@ -506,6 +517,18 @@ impl ExecutorFileSystem for UnsandboxedFileSystem {
         Box::pin(UnsandboxedFileSystem::read_directory(self, path, sandbox))
     }
 
+    fn walk<'a>(
+        &'a self,
+        path: &'a PathUri,
+        options: WalkOptions,
+        sandbox: Option<&'a FileSystemSandboxContext>,
+    ) -> ExecutorFileSystemFuture<'a, WalkOutcome> {
+        Box::pin(async move {
+            reject_platform_sandbox_context(sandbox)?;
+            self.file_system.walk(path, options, /*sandbox*/ None).await
+        })
+    }
+
     fn remove<'a>(
         &'a self,
         path: &'a PathUri,
@@ -677,6 +700,165 @@ impl DirectFileSystem {
         Ok(entries)
     }
 
+    fn sync_walk(
+        root: &PathUri,
+        options: WalkOptions,
+        cancelled: &CancellationToken,
+    ) -> io::Result<WalkOutcome> {
+        if options.max_directories == 0 || options.max_entries == 0 {
+            return Err(io::Error::new(
+                io::ErrorKind::InvalidInput,
+                "filesystem walk limits must be greater than zero",
+            ));
+        }
+        if options.max_depth > MAX_WALK_DEPTH
+            || options.max_directories > MAX_WALK_DIRECTORIES
+            || options.max_entries > MAX_WALK_ENTRIES
+        {
+            return Err(io::Error::new(
+                io::ErrorKind::InvalidInput,
+                format!(
+                    "filesystem walk limits exceed maximums: depth={MAX_WALK_DEPTH}, directories={MAX_WALK_DIRECTORIES}, entries={MAX_WALK_ENTRIES}"
+                ),
+            ));
+        }
+
+        check_walk_cancelled(cancelled)?;
+        let (root_metadata, root_is_symlink) = walk_metadata(root)?;
+        if !root_metadata.is_dir() || (root_is_symlink && !options.follow_directory_symlinks) {
+            return Ok(WalkOutcome::default());
+        }
+
+        let root_identity = if options.follow_directory_symlinks {
+            check_walk_cancelled(cancelled)?;
+            walk_canonicalize(root)?
+        } else {
+            root.clone()
+        };
+        let mut outcome = WalkOutcome::default();
+        let mut queue = VecDeque::from([(root.clone(), 0usize)]);
+        let mut visited_directories = HashSet::from([root_identity]);
+        let mut directory_count = 1usize;
+        let mut entry_count = 0usize;
+        let mut response_bytes = 0usize;
+
+        while let Some((directory, depth)) = queue.pop_front() {
+            let entries = walk_read_directory(&directory, cancelled);
+            check_walk_cancelled(cancelled)?;
+            let mut entries = match entries {
+                Ok(entries) => entries,
+                Err(error) => {
+                    if !push_walk_error(
+                        &mut outcome,
+                        &mut response_bytes,
+                        directory,
+                        error.to_string(),
+                    ) {
+                        return Ok(outcome);
+                    }
+                    continue;
+                }
+            };
+            entries.sort();
+
+            for file_name in entries {
+                check_walk_cancelled(cancelled)?;
+                if entry_count == options.max_entries {
+                    outcome.truncated = true;
+                    return Ok(outcome);
+                }
+                entry_count += 1;
+
+                let path = match directory.join(&file_name) {
+                    Ok(path) => path,
+                    Err(error) => {
+                        if !push_walk_error(
+                            &mut outcome,
+                            &mut response_bytes,
+                            directory.clone(),
+                            error.to_string(),
+                        ) {
+                            return Ok(outcome);
+                        }
+                        continue;
+                    }
+                };
+                let (metadata, is_symlink) = match walk_metadata(&path) {
+                    Ok(metadata) => metadata,
+                    Err(error) => {
+                        if !push_walk_error(
+                            &mut outcome,
+                            &mut response_bytes,
+                            path,
+                            error.to_string(),
+                        ) {
+                            return Ok(outcome);
+                        }
+                        continue;
+                    }
+                };
+                if is_symlink && (!options.follow_directory_symlinks || !metadata.is_dir()) {
+                    continue;
+                }
+
+                let kind = if metadata.is_dir() {
+                    WalkEntryKind::Directory
+                } else if metadata.is_file() {
+                    WalkEntryKind::File
+                } else {
+                    continue;
+                };
+                if !reserve_walk_response_bytes(
+                    &mut outcome,
+                    &mut response_bytes,
+                    path.to_string().len(),
+                ) {
+                    return Ok(outcome);
+                }
+                outcome.entries.push(WalkEntry {
+                    path: path.clone(),
+                    kind,
+                });
+
+                if kind == WalkEntryKind::Directory && depth < options.max_depth {
+                    if options.prune_hidden_directories && file_name.starts_with('.') {
+                        continue;
+                    }
+                    let directory_identity = if options.follow_directory_symlinks {
+                        check_walk_cancelled(cancelled)?;
+                        match walk_canonicalize(&path) {
+                            Ok(path) => path,
+                            Err(error) => {
+                                if !push_walk_error(
+                                    &mut outcome,
+                                    &mut response_bytes,
+                                    path,
+                                    error.to_string(),
+                                ) {
+                                    return Ok(outcome);
+                                }
+                                continue;
+                            }
+                        }
+                    } else {
+                        path.clone()
+                    };
+                    if !visited_directories.insert(directory_identity) {
+                        continue;
+                    }
+                    if directory_count == options.max_directories {
+                        outcome.truncated = true;
+                    } else {
+                        directory_count += 1;
+                        queue.push_back((path, depth + 1));
+                    }
+                }
+            }
+        }
+
+        Ok(outcome)
+    }
+
     async fn remove(
         &self,
         path: &PathUri,
@@ -827,6 +1009,23 @@ impl ExecutorFileSystem for DirectFileSystem {
         Box::pin(DirectFileSystem::read_directory(self, path, sandbox))
     }
 
+    fn walk<'a>(
+        &'a self,
+        path: &'a PathUri,
+        options: WalkOptions,
+        sandbox: Option<&'a FileSystemSandboxContext>,
+    ) -> ExecutorFileSystemFuture<'a, WalkOutcome> {
+        Box::pin(async move {
+            reject_sandbox_context(sandbox)?;
+            let path = path.clone();
+            let cancelled = CancellationToken::new();
+            let _cancel_on_drop = cancelled.clone().drop_guard();
+            tokio::task::spawn_blocking(move || Self::sync_walk(&path, options, &cancelled))
+                .await
+                .map_err(|err| io::Error::other(format!("filesystem task failed: {err}")))?
+        })
+    }
+
     fn remove<'a>(
         &'a self,
         path: &'a PathUri,
@@ -853,6 +1052,89 @@ impl ExecutorFileSystem for DirectFileSystem {
     }
 }
 
+fn check_walk_cancelled(cancelled: &CancellationToken) -> io::Result<()> {
+    if cancelled.is_cancelled() {
+        return Err(io::Error::new(
+            io::ErrorKind::Interrupted,
+            "filesystem walk cancelled",
+        ));
+    }
+    Ok(())
+}
+
+fn walk_metadata(path: &PathUri) -> io::Result<(std::fs::Metadata, bool)> {
+    let path = path.to_abs_path()?;
+    let metadata = std::fs::symlink_metadata(path.as_path())?;
+    let is_symlink = metadata.is_symlink();
+    let metadata = if is_symlink {
+        std::fs::metadata(path.as_path())?
+    } else {
+        metadata
+    };
+    Ok((metadata, is_symlink))
+}
+
+fn walk_canonicalize(path: &PathUri) -> io::Result<PathUri> {
+    let path = path.to_abs_path()?;
+    let canonicalized =
+        AbsolutePathBuf::from_absolute_path(std::fs::canonicalize(path.as_path())?)?;
+    Ok(PathUri::from_abs_path(&canonicalized))
+}
+
+fn walk_read_directory(path: &PathUri, cancelled: &CancellationToken) -> io::Result<Vec<String>> {
+    check_walk_cancelled(cancelled)?;
+    let path = path.to_abs_path()?;
+    let mut entries = Vec::new();
+    for entry in std::fs::read_dir(path.as_path())? {
+        check_walk_cancelled(cancelled)?;
+        let entry = entry?;
+        let Ok(file_type) = entry.file_type() else {
+            continue;
+        };
+        // Match DirectFileSystem::read_directory: omit broken or inaccessible links.
+        if file_type.is_symlink() {
+            check_walk_cancelled(cancelled)?;
+            if std::fs::metadata(entry.path()).is_err() {
+                continue;
+            }
+        }
+        entries.push(entry.file_name().to_string_lossy().into_owned());
+    }
+    Ok(entries)
+}
+
+fn push_walk_error(
+    outcome: &mut WalkOutcome,
+    response_bytes: &mut usize,
+    path: PathUri,
+    message: String,
+) -> bool {
+    let item_bytes = path.to_string().len().saturating_add(message.len());
+    if !reserve_walk_response_bytes(outcome, response_bytes, item_bytes) {
+        return false;
+    }
+    outcome.errors.push(WalkError { path, message });
+    true
+}
+
+fn reserve_walk_response_bytes(
+    outcome: &mut WalkOutcome,
+    response_bytes: &mut usize,
+    content_bytes: usize,
+) -> bool {
+    let item_bytes = content_bytes.saturating_add(WALK_RESPONSE_ITEM_OVERHEAD_BYTES);
+    let Some(total_bytes) = response_bytes.checked_add(item_bytes) else {
+        outcome.truncated = true;
+        return false;
+    };
+    if total_bytes > MAX_WALK_RESPONSE_BYTES {
+        outcome.truncated = true;
+        return false;
+    }
+    *response_bytes = total_bytes;
+    true
+}
+
 fn reject_sandbox_context(sandbox: Option<&FileSystemSandboxContext>) -> io::Result<()> {
     if sandbox.is_some() {
         return Err(io::Error::new(
@@ -1039,3 +1321,103 @@ mod tests {
         Ok(())
     }
 }
+
+#[cfg(test)]
+mod walk_tests {
+    use super::*;
+    use codex_protocol::models::PermissionProfile;
+    use codex_protocol::permissions::FileSystemSandboxPolicy;
+    use codex_protocol::permissions::NetworkSandboxPolicy;
+    use pretty_assertions::assert_eq;
+
+    #[tokio::test]
+    async fn sync_walk_rejects_sandbox_context() -> io::Result<()> {
+        let temp = tempfile::tempdir()?;
+        let root = PathUri::from_host_native_path(temp.path())?;
+        let sandbox = FileSystemSandboxContext::from_permission_profile(
+            PermissionProfile::from_runtime_permissions(
+                &FileSystemSandboxPolicy::restricted(Vec::new()),
+                NetworkSandboxPolicy::Restricted,
+            ),
+        );
+        let options = WalkOptions {
+            max_depth: 1,
+            max_directories: 1,
+            max_entries: 1,
+            follow_directory_symlinks: false,
+            prune_hidden_directories: false,
+        };
+        let direct_error = DirectFileSystem
+            .walk(&root, options, Some(&sandbox))
+            .await
+            .expect_err("direct walk must reject sandbox contexts");
+        let wrapper_error = UnsandboxedFileSystem::default()
+            .walk(&root, options, Some(&sandbox))
+            .await
+            .expect_err("unsandboxed walk must reject restricted contexts");
+        assert_eq!(direct_error.kind(), io::ErrorKind::InvalidInput);
+        assert_eq!(wrapper_error.kind(), io::ErrorKind::InvalidInput);
+        Ok(())
+    }
+
+    #[test]
+    fn sync_walk_cancellation_stops_before_io() -> io::Result<()> {
+        let temp = tempfile::tempdir()?;
+        let missing = PathUri::from_host_native_path(temp.path().join("missing"))?;
+        let options = WalkOptions {
+            max_depth: 1,
+            max_directories: 1,
+            max_entries: 1,
+            follow_directory_symlinks: true,
+            prune_hidden_directories: false,
+        };
+        let cancelled = CancellationToken::new();
+        let cancel_on_drop = cancelled.clone().drop_guard();
+        drop(cancel_on_drop);
+
+        for result in [
+            DirectFileSystem::sync_walk(&missing, options, &cancelled).map(|_| ()),
+            walk_read_directory(&missing, &cancelled).map(|_| ()),
+        ] {
+            assert_eq!(
+                result
+                    .expect_err("cancelled walks must stop before I/O")
+                    .kind(),
+                io::ErrorKind::Interrupted,
+            );
+        }
+        Ok(())
+    }
+
+    #[test]
+    fn sync_walk_response_budget_counts_entries_and_errors() -> io::Result<()> {
+        let temp = tempfile::tempdir()?;
+        let root = PathUri::from_host_native_path(temp.path())?;
+        let mut outcome = WalkOutcome::default();
+        let mut response_bytes =
+            MAX_WALK_RESPONSE_BYTES - WALK_RESPONSE_ITEM_OVERHEAD_BYTES - root.to_string().len();
+        assert!(push_walk_error(
+            &mut outcome,
+            &mut response_bytes,
+            root.clone(),
+            String::new()
+        ));
+        assert!(!reserve_walk_response_bytes(
+            &mut outcome,
+            &mut response_bytes,
+            /*content_bytes*/ 0
+        ));
+        assert_eq!(
+            outcome,
+            WalkOutcome {
+                entries: Vec::new(),
+                errors: vec![WalkError {
+                    path: root,
+                    message: String::new()
+                }],
+                truncated: true,
+            },
+        );
+        Ok(())
+    }
+}

codex-rs/exec-server/src/remote_file_system.rs

@@ -37,7 +37,6 @@ use crate::protocol::FsWalkParams;
 use crate::protocol::FsWriteFileParams;
 
 const INVALID_REQUEST_ERROR_CODE: i64 = -32600;
-const METHOD_NOT_FOUND_ERROR_CODE: i64 = -32601;
 const NOT_FOUND_ERROR_CODE: i64 = -32004;
 
 #[path = "remote_file_stream.rs"]
@@ -252,27 +251,14 @@ impl RemoteFileSystem {
     ) -> FileSystemResult<WalkOutcome> {
         trace!("remote fs walk");
         let client = self.client.get().await.map_err(map_remote_error)?;
-        let response = match client
+        client
             .fs_walk(FsWalkParams {
                 path: path.clone(),
                 options,
                 sandbox: remote_sandbox_context(sandbox),
             })
             .await
-        {
-            Ok(response) => response,
-            Err(ExecServerError::Server {
-                code: METHOD_NOT_FOUND_ERROR_CODE,
-                ..
-            }) => {
-                return <Self as ExecutorFileSystem>::walk_via_directory_reads(
-                    self, path, options, sandbox,
-                )
-                .await;
-            }
-            Err(error) => return Err(map_remote_error(error)),
-        };
-        Ok(response)
+            .map_err(map_remote_error)
     }
 
     async fn remove(