[codex] exec-server: stream files in chunks

codex-rs/app-server/tests/suite/v2/external_agent_config.rs

@@ -886,24 +886,14 @@ async fn external_agent_config_import_returns_before_background_session_import_f
     let session_path = session_dir.join("session.jsonl");
     std::fs::create_dir_all(&project_root)?;
     std::fs::create_dir_all(&session_dir)?;
-    std::fs::write(
-        &session_path,
-        serde_json::json!({
-            "type": "user",
-            "cwd": &project_root,
-            "timestamp": &recent_timestamp,
-            "message": { "content": "first request" },
-        })
-        .to_string(),
-    )?;
-
-    let project_config_dir = project_root.join(".codex");
-    std::fs::create_dir_all(&project_config_dir)?;
-    let project_config = project_config_dir.join("config.toml");
-    let status = std::process::Command::new("mkfifo")
-        .arg(&project_config)
-        .status()?;
-    assert!(status.success());
+    let session_contents = serde_json::json!({
+        "type": "user",
+        "cwd": &project_root,
+        "timestamp": &recent_timestamp,
+        "message": { "content": "first request" },
+    })
+    .to_string();
+    std::fs::write(&session_path, &session_contents)?;
 
     let home_dir = codex_home.path().display().to_string();
     let mut mcp =
@@ -926,6 +916,12 @@ async fn external_agent_config_import_returns_before_background_session_import_f
     assert_eq!(detected.items.len(), 1);
     let detected_items = detected.items;
 
+    std::fs::remove_file(&session_path)?;
+    let status = std::process::Command::new("mkfifo")
+        .arg(&session_path)
+        .status()?;
+    assert!(status.success());
+
     let request_id = mcp
         .send_raw_request(
             "externalAgentConfig/import",
@@ -964,17 +960,17 @@ async fn external_agent_config_import_returns_before_background_session_import_f
     let response: ExternalAgentConfigImportResponse = to_response(response)?;
     let duplicate_import_id = assert_import_response(response);
 
-    let writer = tokio::spawn(async move {
-        let mut file = tokio::fs::OpenOptions::new()
-            .write(true)
-            .open(&project_config)
-            .await?;
-        file.write_all(b"\n").await
-    });
-    timeout(DEFAULT_TIMEOUT, writer).await???;
-
     let mut completed_import_ids = Vec::new();
     for _ in 0..2 {
+        timeout(DEFAULT_TIMEOUT, async {
+            let mut file = tokio::fs::OpenOptions::new()
+                .write(true)
+                .open(&session_path)
+                .await?;
+            file.write_all(session_contents.as_bytes()).await
+        })
+        .await??;
+
         let notification = timeout(
             DEFAULT_TIMEOUT,
             mcp.read_stream_until_notification_message("externalAgentConfig/import/completed"),

codex-rs/config/src/loader/tests.rs

@@ -3,6 +3,7 @@ use codex_file_system::CopyOptions;
 use codex_file_system::CreateDirectoryOptions;
 use codex_file_system::ExecutorFileSystemFuture;
 use codex_file_system::FileMetadata;
+use codex_file_system::FileSystemReadStream;
 use codex_file_system::FileSystemSandboxContext;
 use codex_file_system::ReadDirectoryEntry;
 use codex_file_system::RemoveOptions;
@@ -36,6 +37,19 @@ impl ExecutorFileSystem for TestFileSystem {
         })
     }
 
+    fn read_file_stream<'a>(
+        &'a self,
+        _path: &'a PathUri,
+        _sandbox: Option<&'a FileSystemSandboxContext>,
+    ) -> ExecutorFileSystemFuture<'a, FileSystemReadStream> {
+        Box::pin(async {
+            Err(std::io::Error::new(
+                std::io::ErrorKind::Unsupported,
+                "test filesystem does not support streaming reads",
+            ))
+        })
+    }
+
     fn write_file<'a>(
         &'a self,
         _path: &'a PathUri,

codex-rs/core-plugins/src/provider_tests.rs

@@ -8,6 +8,7 @@ use codex_exec_server::EnvironmentManager;
 use codex_exec_server::ExecutorFileSystem;
 use codex_exec_server::ExecutorFileSystemFuture;
 use codex_exec_server::FileMetadata;
+use codex_exec_server::FileSystemReadStream;
 use codex_exec_server::FileSystemResult;
 use codex_exec_server::FileSystemSandboxContext;
 use codex_exec_server::LOCAL_ENVIRONMENT_ID;
@@ -89,6 +90,14 @@ impl ExecutorFileSystem for SyntheticPluginFileSystem {
         })
     }
 
+    fn read_file_stream<'a>(
+        &'a self,
+        _path: &'a PathUri,
+        _sandbox: Option<&'a FileSystemSandboxContext>,
+    ) -> ExecutorFileSystemFuture<'a, FileSystemReadStream> {
+        Box::pin(async { Self::unsupported() })
+    }
+
     fn write_file<'a>(
         &'a self,
         _path: &'a PathUri,

codex-rs/core/src/agents_md_tests.rs

@@ -11,6 +11,7 @@ use codex_exec_server::CreateDirectoryOptions;
 use codex_exec_server::Environment;
 use codex_exec_server::ExecutorFileSystemFuture;
 use codex_exec_server::FileMetadata;
+use codex_exec_server::FileSystemReadStream;
 use codex_exec_server::FileSystemSandboxContext;
 use codex_exec_server::LOCAL_FS;
 use codex_exec_server::ReadDirectoryEntry;
@@ -140,6 +141,19 @@ impl ExecutorFileSystem for FailingFileSystem {
         Box::pin(FailingFileSystem::read_file(self, path, sandbox))
     }
 
+    fn read_file_stream<'a>(
+        &'a self,
+        _path: &'a PathUri,
+        _sandbox: Option<&'a FileSystemSandboxContext>,
+    ) -> ExecutorFileSystemFuture<'a, FileSystemReadStream> {
+        Box::pin(async {
+            Err(io::Error::new(
+                io::ErrorKind::Unsupported,
+                "failing filesystem does not support streaming reads",
+            ))
+        })
+    }
+
     fn write_file<'a>(
         &'a self,
         path: &'a PathUri,

codex-rs/exec-server/Cargo.toml

@@ -45,11 +45,20 @@ tokio = { workspace = true, features = [
     "sync",
     "time",
 ] }
-tokio-util = { workspace = true, features = ["rt"] }
+tokio-util = { workspace = true, features = ["io", "rt"] }
 tokio-tungstenite = { workspace = true }
 tracing = { workspace = true }
 uuid = { workspace = true, features = ["v4"] }
 
+[target.'cfg(unix)'.dependencies]
+libc = { workspace = true }
+
+[target.'cfg(windows)'.dependencies]
+windows-sys = { version = "0.52", features = [
+    "Win32_Foundation",
+    "Win32_Storage_FileSystem",
+] }
+
 [dev-dependencies]
 anyhow = { workspace = true }
 codex-test-binary-support = { workspace = true }

codex-rs/exec-server/README.md

@@ -345,6 +345,8 @@ invalid or unavailable paths. For compatibility, requests also accept native
 absolute path strings and normalize them to `file:` URIs:
 
 - `fs/readFile`
+- `fs/open`, `fs/readBlock`, and `fs/close` (internal transport for
+  `ExecutorFileSystem::read_file_stream`)
 - `fs/writeFile`
 - `fs/createDirectory`
 - `fs/getMetadata`

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

@@ -45,21 +45,30 @@ use crate::protocol::ExecOutputDeltaNotification;
 use crate::protocol::ExecParams;
 use crate::protocol::ExecResponse;
 use crate::protocol::FS_CANONICALIZE_METHOD;
+use crate::protocol::FS_CLOSE_METHOD;
 use crate::protocol::FS_COPY_METHOD;
 use crate::protocol::FS_CREATE_DIRECTORY_METHOD;
 use crate::protocol::FS_GET_METADATA_METHOD;
+use crate::protocol::FS_OPEN_METHOD;
+use crate::protocol::FS_READ_BLOCK_METHOD;
 use crate::protocol::FS_READ_DIRECTORY_METHOD;
 use crate::protocol::FS_READ_FILE_METHOD;
 use crate::protocol::FS_REMOVE_METHOD;
 use crate::protocol::FS_WRITE_FILE_METHOD;
 use crate::protocol::FsCanonicalizeParams;
 use crate::protocol::FsCanonicalizeResponse;
+use crate::protocol::FsCloseParams;
+use crate::protocol::FsCloseResponse;
 use crate::protocol::FsCopyParams;
 use crate::protocol::FsCopyResponse;
 use crate::protocol::FsCreateDirectoryParams;
 use crate::protocol::FsCreateDirectoryResponse;
 use crate::protocol::FsGetMetadataParams;
 use crate::protocol::FsGetMetadataResponse;
+use crate::protocol::FsOpenParams;
+use crate::protocol::FsOpenResponse;
+use crate::protocol::FsReadBlockParams;
+use crate::protocol::FsReadBlockResponse;
 use crate::protocol::FsReadDirectoryParams;
 use crate::protocol::FsReadDirectoryResponse;
 use crate::protocol::FsReadFileParams;
@@ -430,6 +439,24 @@ impl ExecServerClient {
         self.call(FS_READ_FILE_METHOD, &params).await
     }
 
+    pub async fn fs_open(&self, params: FsOpenParams) -> Result<FsOpenResponse, ExecServerError> {
+        self.call(FS_OPEN_METHOD, &params).await
+    }
+
+    pub async fn fs_read_block(
+        &self,
+        params: FsReadBlockParams,
+    ) -> Result<FsReadBlockResponse, ExecServerError> {
+        self.call(FS_READ_BLOCK_METHOD, &params).await
+    }
+
+    pub async fn fs_close(
+        &self,
+        params: FsCloseParams,
+    ) -> Result<FsCloseResponse, ExecServerError> {
+        self.call(FS_CLOSE_METHOD, &params).await
+    }
+
     pub async fn fs_write_file(
         &self,
         params: FsWriteFileParams,

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

@@ -0,0 +1,128 @@
+use std::collections::HashMap;
+use std::fs::File;
+use std::io;
+use std::sync::Arc;
+
+use codex_file_system::FILE_READ_CHUNK_SIZE;
+use tokio::sync::Mutex;
+
+const MAX_OPEN_FILE_READS: usize = 128;
+
+#[derive(Debug, Eq, PartialEq)]
+pub(crate) struct FileReadBlock {
+    pub(crate) bytes: Vec<u8>,
+    pub(crate) eof: bool,
+}
+
+#[derive(Clone, Default)]
+pub(crate) struct FileReadHandleManager {
+    handles: Arc<Mutex<HashMap<String, Arc<File>>>>,
+}
+
+impl FileReadHandleManager {
+    pub(crate) async fn open(
+        &self,
+        handle_id: String,
+        file: tokio::fs::File,
+    ) -> io::Result<String> {
+        let file = Arc::new(file.into_std().await);
+        let mut handles = self.handles.lock().await;
+        if handles.contains_key(&handle_id) {
+            return Err(io::Error::new(
+                io::ErrorKind::InvalidInput,
+                format!("file read handle `{handle_id}` already exists"),
+            ));
+        }
+        if handles.len() >= MAX_OPEN_FILE_READS {
+            return Err(io::Error::new(
+                io::ErrorKind::InvalidInput,
+                format!("at most {MAX_OPEN_FILE_READS} file reads may be open per connection"),
+            ));
+        }
+        handles.insert(handle_id.clone(), file);
+        Ok(handle_id)
+    }
+
+    pub(crate) async fn read_block(
+        &self,
+        handle_id: &str,
+        offset: u64,
+        len: usize,
+    ) -> io::Result<FileReadBlock> {
+        validate_read_block_len(len)?;
+        let file = {
+            let handles = self.handles.lock().await;
+            handles
+                .get(handle_id)
+                .cloned()
+                .ok_or_else(|| unknown_handle_error(handle_id))?
+        };
+        let result =
+            match tokio::task::spawn_blocking(move || read_block_at(&file, offset, len)).await {
+                Ok(result) => result,
+                Err(error) => Err(io::Error::other(format!(
+                    "file read task stopped unexpectedly: {error}"
+                ))),
+            };
+        if result.is_err() {
+            self.close(handle_id).await;
+        }
+        result
+    }
+
+    pub(crate) async fn close(&self, handle_id: &str) {
+        self.handles.lock().await.remove(handle_id);
+    }
+
+    pub(crate) async fn close_all(&self) {
+        self.handles.lock().await.clear();
+    }
+}
+
+fn read_block_at(file: &File, offset: u64, len: usize) -> io::Result<FileReadBlock> {
+    let mut bytes = vec![0; len];
+    let mut bytes_read = 0;
+    while bytes_read < len {
+        let read_offset = offset.checked_add(bytes_read as u64).ok_or_else(|| {
+            io::Error::new(io::ErrorKind::InvalidInput, "file read offset overflowed")
+        })?;
+        match read_file_at(file, &mut bytes[bytes_read..], read_offset) {
+            Ok(0) => break,
+            Ok(read) => bytes_read += read,
+            Err(error) if error.kind() == io::ErrorKind::Interrupted => {}
+            Err(error) => return Err(error),
+        }
+    }
+    bytes.truncate(bytes_read);
+    Ok(FileReadBlock {
+        eof: bytes_read < len,
+        bytes,
+    })
+}
+
+#[cfg(unix)]
+fn read_file_at(file: &File, bytes: &mut [u8], offset: u64) -> io::Result<usize> {
+    std::os::unix::fs::FileExt::read_at(file, bytes, offset)
+}
+
+#[cfg(windows)]
+fn read_file_at(file: &File, bytes: &mut [u8], offset: u64) -> io::Result<usize> {
+    std::os::windows::fs::FileExt::seek_read(file, bytes, offset)
+}
+
+fn validate_read_block_len(len: usize) -> io::Result<()> {
+    if !(1..=FILE_READ_CHUNK_SIZE).contains(&len) {
+        return Err(io::Error::new(
+            io::ErrorKind::InvalidInput,
+            format!("file read block length must be between 1 and {FILE_READ_CHUNK_SIZE}"),
+        ));
+    }
+    Ok(())
+}
+
+fn unknown_handle_error(handle_id: &str) -> io::Error {
+    io::Error::new(
+        io::ErrorKind::NotFound,
+        format!("unknown file read handle `{handle_id}`"),
+    )
+}