Add a bounded filesystem walk RPC (#29841)

Stack 1 of 3. Follow-ups: #29842 and #29844.

## What changes

Adds a general bounded `fs/walk` operation to the exec server.

The operation returns file and directory entries plus recoverable
per-path errors. It skips symlinks, preserves the existing filesystem
sandbox routing, and enforces depth, directory, entry, and response-size
limits.

This PR only defines and wires the filesystem operation. It does not
change any callers yet.

codex-rs/file-system/src/lib.rs

@@ -12,6 +12,7 @@ use codex_protocol::protocol::SandboxPolicy;
 use codex_utils_absolute_path::AbsolutePathBuf;
 use codex_utils_path_uri::PathUri;
 use futures::Stream;
+use std::collections::VecDeque;
 use std::future::Future;
 use std::io;
 use std::path::Path;
@@ -21,6 +22,11 @@ use std::task::Poll;
 
 /// Maximum chunk size returned by [`ExecutorFileSystem::read_file_stream`].
 pub const FILE_READ_CHUNK_SIZE: usize = 1024 * 1024;
+const MAX_WALK_DEPTH: usize = 64;
+const MAX_WALK_DIRECTORIES: usize = 10_000;
+const MAX_WALK_ENTRIES: usize = 50_000;
+const MAX_WALK_RESPONSE_BYTES: usize = 4 * 1024 * 1024;
+const WALK_RESPONSE_ITEM_OVERHEAD_BYTES: usize = 64;
 
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
 pub struct CreateDirectoryOptions {
@@ -56,6 +62,51 @@ pub struct ReadDirectoryEntry {
     pub is_file: bool,
 }
 
+/// Bounds for a recursive filesystem walk.
+#[derive(Clone, Copy, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct WalkOptions {
+    /// Maximum directory depth below the root that may be traversed.
+    pub max_depth: usize,
+    /// Maximum number of directories that may be traversed, including the root.
+    pub max_directories: usize,
+    /// Maximum number of directory entries that may be examined.
+    pub max_entries: usize,
+}
+
+/// Type of a filesystem entry returned by a walk.
+#[derive(Clone, Copy, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)]
+#[serde(rename_all = "camelCase")]
+pub enum WalkEntryKind {
+    Directory,
+    File,
+}
+
+/// One non-symlink entry returned by a walk.
+#[derive(Clone, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct WalkEntry {
+    pub path: PathUri,
+    pub kind: WalkEntryKind,
+}
+
+/// A descendant that could not be inspected during a walk.
+#[derive(Clone, Debug, Eq, PartialEq, serde::Deserialize, serde::Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct WalkError {
+    pub path: PathUri,
+    pub message: String,
+}
+
+/// Entries and recoverable errors collected by a bounded walk.
+#[derive(Clone, Debug, Default, Eq, PartialEq, serde::Deserialize, serde::Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct WalkOutcome {
+    pub entries: Vec<WalkEntry>,
+    pub errors: Vec<WalkError>,
+    pub truncated: bool,
+}
+
 #[derive(Clone, Debug, Eq, PartialEq, serde::Serialize, serde::Deserialize)]
 #[serde(rename_all = "camelCase")]
 pub struct FileSystemSandboxContext {
@@ -249,6 +300,16 @@ pub trait ExecutorFileSystem: Send + Sync {
         sandbox: Option<&'a FileSystemSandboxContext>,
     ) -> ExecutorFileSystemFuture<'a, Vec<ReadDirectoryEntry>>;
 
+    /// Recursively lists descendants, skipping symlinks.
+    fn walk<'a>(
+        &'a self,
+        path: &'a PathUri,
+        options: WalkOptions,
+        sandbox: Option<&'a FileSystemSandboxContext>,
+    ) -> ExecutorFileSystemFuture<'a, WalkOutcome> {
+        Box::pin(walk(self, path, options, sandbox))
+    }
+
     fn remove<'a>(
         &'a self,
         path: &'a PathUri,
@@ -264,3 +325,155 @@ pub trait ExecutorFileSystem: Send + Sync {
         sandbox: Option<&'a FileSystemSandboxContext>,
     ) -> ExecutorFileSystemFuture<'a, ()>;
 }
+
+async fn walk<F: ExecutorFileSystem + ?Sized>(
+    file_system: &F,
+    root: &PathUri,
+    options: WalkOptions,
+    sandbox: Option<&FileSystemSandboxContext>,
+) -> FileSystemResult<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}"
+            ),
+        ));
+    }
+
+    let root_metadata = file_system.get_metadata(root, sandbox).await?;
+    if root_metadata.is_symlink || !root_metadata.is_directory {
+        return Ok(WalkOutcome::default());
+    }
+
+    let mut outcome = WalkOutcome::default();
+    let mut queue = VecDeque::from([(root.clone(), 0usize)]);
+    let mut directory_count = 1usize;
+    let mut entry_count = 0usize;
+    let mut response_bytes = 0usize;
+
+    while let Some((directory, depth)) = queue.pop_front() {
+        let mut entries = match file_system.read_directory(&directory, sandbox).await {
+            Ok(entries) => entries,
+            Err(error) => {
+                if !push_walk_error(
+                    &mut outcome,
+                    &mut response_bytes,
+                    directory,
+                    error.to_string(),
+                ) {
+                    return Ok(outcome);
+                }
+                continue;
+            }
+        };
+        entries.sort_by(|left, right| left.file_name.cmp(&right.file_name));
+
+        for entry in entries {
+            if entry_count == options.max_entries {
+                outcome.truncated = true;
+                return Ok(outcome);
+            }
+            entry_count += 1;
+
+            let path = match directory.join(&entry.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 = match file_system.get_metadata(&path, sandbox).await {
+                Ok(metadata) => metadata,
+                Err(error) => {
+                    if !push_walk_error(&mut outcome, &mut response_bytes, path, error.to_string())
+                    {
+                        return Ok(outcome);
+                    }
+                    continue;
+                }
+            };
+            if metadata.is_symlink {
+                continue;
+            }
+
+            let kind = if metadata.is_directory {
+                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 directory_count == options.max_directories {
+                    outcome.truncated = true;
+                } else {
+                    directory_count += 1;
+                    queue.push_back((path, depth + 1));
+                }
+            }
+        }
+    }
+
+    Ok(outcome)
+}
+
+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
+}