diff --git a/codex-rs/exec-server/src/client.rs b/codex-rs/exec-server/src/client.rs index b6c56e9f5e..458693743f 100644 --- a/codex-rs/exec-server/src/client.rs +++ b/codex-rs/exec-server/src/client.rs @@ -214,6 +214,10 @@ impl LazyRemoteExecServerClient { } } + pub(crate) fn new_connection_client(&self) -> Self { + Self::new(self.transport_params.clone()) + } + pub(crate) async fn get(&self) -> Result { if let Some(client) = self.connected_client() { return Ok(client); diff --git a/codex-rs/exec-server/src/remote_file_system.rs b/codex-rs/exec-server/src/remote_file_system.rs index 8c3a9ebbe8..b0b39aea37 100644 --- a/codex-rs/exec-server/src/remote_file_system.rs +++ b/codex-rs/exec-server/src/remote_file_system.rs @@ -4,8 +4,11 @@ use base64::engine::general_purpose::STANDARD; use codex_utils_absolute_path::AbsolutePathBuf; use std::future::Future; use std::path::Path; +use std::sync::OnceLock; use std::thread; use tokio::io; +use tokio::runtime::Builder; +use tokio::runtime::Runtime; use tracing::trace; use crate::CopyOptions; @@ -33,15 +36,20 @@ use crate::protocol::FsWriteFileParams; const INVALID_REQUEST_ERROR_CODE: i64 = -32600; const NOT_FOUND_ERROR_CODE: i64 = -32004; -#[derive(Clone)] pub(crate) struct RemoteFileSystem { - client: LazyRemoteExecServerClient, + async_client: LazyRemoteExecServerClient, + sync_client: LazyRemoteExecServerClient, + sync_call_runtime: SyncCallRuntime, } impl RemoteFileSystem { pub(crate) fn new(client: LazyRemoteExecServerClient) -> Self { trace!("remote fs new"); - Self { client } + Self { + async_client: client, + sync_client: client.new_connection_client(), + sync_call_runtime: SyncCallRuntime::default(), + } } fn run_sync_remote_call( @@ -54,24 +62,55 @@ impl RemoteFileSystem { F: FnOnce(ExecServerClient) -> Fut + Send + 'static, Fut: Future> + Send + 'static, { - let runtime = tokio::runtime::Handle::try_current().map_err(|err| { - io::Error::other(format!("remote fs {operation} requires runtime: {err}")) - })?; - let client = self.client.clone(); + let runtime = self.sync_call_runtime.get()?; + let client = self.sync_client.clone(); // ExecutorFileSystem exposes synchronous path operations, so remote callers have to hop - // through the active runtime from a helper thread to reach the JSON-RPC filesystem - // surface without blocking the runtime worker. - thread::spawn(move || { - runtime.block_on(async move { - let client = client.get().await.map_err(map_remote_error)?; - call(client).await - }) + // through a dedicated exec-server client/runtime from a helper thread. Reusing the caller + // runtime can deadlock current-thread runtimes because the JSON-RPC reader also needs that + // runtime to make progress. + thread::scope(|scope| { + scope + .spawn(move || { + runtime.block_on(async move { + let client = client.get().await.map_err(map_remote_error)?; + call(client).await + }) + }) + .join() }) - .join() .map_err(|_| io::Error::other(format!("remote fs {operation} thread panicked")))? } } +#[derive(Default)] +struct SyncCallRuntime { + runtime: OnceLock>, +} + +impl SyncCallRuntime { + fn get(&self) -> FileSystemResult<&Runtime> { + self.runtime + .get_or_init(|| { + Builder::new_multi_thread() + .worker_threads(1) + .thread_name("codex-remote-fs") + .enable_all() + .build() + .map_err(|err| err.to_string()) + }) + .as_ref() + .map_err(|err| io::Error::other(format!("failed to start remote fs runtime: {err}"))) + } +} + +impl Drop for SyncCallRuntime { + fn drop(&mut self) { + if let Some(Ok(runtime)) = self.runtime.take() { + runtime.shutdown_background(); + } + } +} + #[async_trait] impl ExecutorFileSystem for RemoteFileSystem { fn canonicalize(&self, path: &AbsolutePathBuf) -> FileSystemResult { @@ -124,7 +163,7 @@ impl ExecutorFileSystem for RemoteFileSystem { sandbox: Option<&FileSystemSandboxContext>, ) -> FileSystemResult> { trace!("remote fs read_file"); - let client = self.client.get().await.map_err(map_remote_error)?; + let client = self.async_client.get().await.map_err(map_remote_error)?; let response = client .fs_read_file(FsReadFileParams { path: path.clone(), @@ -147,7 +186,7 @@ impl ExecutorFileSystem for RemoteFileSystem { sandbox: Option<&FileSystemSandboxContext>, ) -> FileSystemResult<()> { trace!("remote fs write_file"); - let client = self.client.get().await.map_err(map_remote_error)?; + let client = self.async_client.get().await.map_err(map_remote_error)?; client .fs_write_file(FsWriteFileParams { path: path.clone(), @@ -166,7 +205,7 @@ impl ExecutorFileSystem for RemoteFileSystem { sandbox: Option<&FileSystemSandboxContext>, ) -> FileSystemResult<()> { trace!("remote fs create_directory"); - let client = self.client.get().await.map_err(map_remote_error)?; + let client = self.async_client.get().await.map_err(map_remote_error)?; client .fs_create_directory(FsCreateDirectoryParams { path: path.clone(), @@ -184,7 +223,7 @@ impl ExecutorFileSystem for RemoteFileSystem { sandbox: Option<&FileSystemSandboxContext>, ) -> FileSystemResult { trace!("remote fs get_metadata"); - let client = self.client.get().await.map_err(map_remote_error)?; + let client = self.async_client.get().await.map_err(map_remote_error)?; let response = client .fs_get_metadata(FsGetMetadataParams { path: path.clone(), @@ -207,7 +246,7 @@ impl ExecutorFileSystem for RemoteFileSystem { sandbox: Option<&FileSystemSandboxContext>, ) -> FileSystemResult> { trace!("remote fs read_directory"); - let client = self.client.get().await.map_err(map_remote_error)?; + let client = self.async_client.get().await.map_err(map_remote_error)?; let response = client .fs_read_directory(FsReadDirectoryParams { path: path.clone(), @@ -233,7 +272,7 @@ impl ExecutorFileSystem for RemoteFileSystem { sandbox: Option<&FileSystemSandboxContext>, ) -> FileSystemResult<()> { trace!("remote fs remove"); - let client = self.client.get().await.map_err(map_remote_error)?; + let client = self.async_client.get().await.map_err(map_remote_error)?; client .fs_remove(FsRemoveParams { path: path.clone(), @@ -254,7 +293,7 @@ impl ExecutorFileSystem for RemoteFileSystem { sandbox: Option<&FileSystemSandboxContext>, ) -> FileSystemResult<()> { trace!("remote fs copy"); - let client = self.client.get().await.map_err(map_remote_error)?; + let client = self.async_client.get().await.map_err(map_remote_error)?; client .fs_copy(FsCopyParams { source_path: source_path.clone(),