From 6dfbcb8327cc404180a1775700ef83c74d0990e1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=90=AC=E9=A3=8E=E5=90=B9=E9=9B=A8?= <2686367411@qq.com> Date: Tue, 10 Nov 2026 23:33:07 -0800 Subject: [PATCH] fix(acp): add timeout for filesystem semaphore acquisition to prevent agent hangs --- src-tauri/src/acp/file_system_runtime.rs | 134 ++++++++++++++++++++--- 1 file changed, 117 insertions(+), 17 deletions(-) diff --git a/src-tauri/src/acp/file_system_runtime.rs b/src-tauri/src/acp/file_system_runtime.rs index 95f2f5d87e..564f30ec64 100644 --- a/src-tauri/src/acp/file_system_runtime.rs +++ b/src-tauri/src/acp/file_system_runtime.rs @@ -9,12 +9,13 @@ use std::time::{Duration, Instant}; use agent_client_protocol::schema::v1::{ ReadTextFileRequest, ReadTextFileResponse, WriteTextFileRequest, WriteTextFileResponse, }; -use tokio::sync::Semaphore; +use tokio::sync::{OwnedSemaphorePermit, Semaphore}; use crate::models::agent::AgentType; use crate::parsers::expand_home_prefix; const FS_MAX_CONCURRENT_OPS: usize = 8; +const FS_ACQUIRE_TIMEOUT: Duration = Duration::from_secs(15); const FS_IO_TIMEOUT: Duration = Duration::from_secs(30); const FS_MAX_FILE_SIZE_BYTES: u64 = 16 * 1024 * 1024; const FS_MAX_READ_RESPONSE_BYTES: usize = 2 * 1024 * 1024; @@ -978,6 +979,23 @@ impl FileSystemRuntime { } } + async fn acquire_io_permit( + &self, + operation: &'static str, + ) -> Result { + tokio::time::timeout(FS_ACQUIRE_TIMEOUT, self.io_semaphore.clone().acquire_owned()) + .await + .map_err(|_| { + FileSystemRuntimeError::Internal(format!( + "{operation} timed out waiting for filesystem permit after {}s", + FS_ACQUIRE_TIMEOUT.as_secs() + )) + })? + .map_err(|_| { + FileSystemRuntimeError::Internal("filesystem runtime closed".to_string()) + }) + } + pub async fn read_text_file( &self, request: ReadTextFileRequest, @@ -993,14 +1011,7 @@ impl FileSystemRuntime { )); } - let _permit = self - .io_semaphore - .clone() - .acquire_owned() - .await - .map_err(|_| { - FileSystemRuntimeError::Internal("filesystem runtime closed".to_string()) - })?; + let _permit = self.acquire_io_permit("fs/read_text_file").await?; let policy = self.policy.clone(); let path = request.path; @@ -1035,14 +1046,7 @@ impl FileSystemRuntime { ))); } - let _permit = self - .io_semaphore - .clone() - .acquire_owned() - .await - .map_err(|_| { - FileSystemRuntimeError::Internal("filesystem runtime closed".to_string()) - })?; + let _permit = self.acquire_io_permit("fs/write_text_file").await?; let policy = self.policy.clone(); let path = request.path; @@ -3174,4 +3178,100 @@ mod tests { let _ = fs::remove_dir_all(workspace); } + + #[tokio::test(start_paused = true)] + async fn read_times_out_when_permits_exhausted() { + let workspace = temp_workspace(); + let file = workspace.join("sample.txt"); + fs::write(&file, "hello").expect("write file"); + + let runtime = FileSystemRuntime::new(workspace.clone()); + let _permits: Vec<_> = (0..FS_MAX_CONCURRENT_OPS) + .map(|_| { + runtime + .io_semaphore + .clone() + .try_acquire_owned() + .expect("acquire permit") + }) + .collect(); + + let err = runtime + .read_text_file(ReadTextFileRequest::new("sid", &file)) + .await + .expect_err("read should time out when permits are exhausted"); + + match &err { + FileSystemRuntimeError::Internal(msg) => { + assert!( + msg.contains("fs/read_text_file timed out waiting for filesystem permit after 15s"), + "unexpected error message: {msg}" + ); + } + other => panic!("expected Internal timeout error, got: {other:?}"), + } + + let rpc_err = err.into_rpc_error(); + assert_eq!( + rpc_err.code, + agent_client_protocol::schema::v1::ErrorCode::InternalError + ); + + drop(_permits); + let ok = runtime + .read_text_file(ReadTextFileRequest::new("sid", &file)) + .await + .expect("read should succeed after permits released"); + assert_eq!(ok.content, "hello"); + + let _ = fs::remove_dir_all(workspace); + } + + #[tokio::test(start_paused = true)] + async fn write_times_out_when_permits_exhausted() { + let workspace = temp_workspace(); + let file = workspace.join("sample.txt"); + + let runtime = FileSystemRuntime::new(workspace.clone()); + let _permits: Vec<_> = (0..FS_MAX_CONCURRENT_OPS) + .map(|_| { + runtime + .io_semaphore + .clone() + .try_acquire_owned() + .expect("acquire permit") + }) + .collect(); + + let err = runtime + .write_text_file(WriteTextFileRequest::new("sid", &file, "world")) + .await + .expect_err("write should time out when permits are exhausted"); + + match &err { + FileSystemRuntimeError::Internal(msg) => { + assert!( + msg.contains("fs/write_text_file timed out waiting for filesystem permit after 15s"), + "unexpected error message: {msg}" + ); + } + other => panic!("expected Internal timeout error, got: {other:?}"), + } + + let rpc_err = err.into_rpc_error(); + assert_eq!( + rpc_err.code, + agent_client_protocol::schema::v1::ErrorCode::InternalError + ); + + drop(_permits); + runtime + .write_text_file(WriteTextFileRequest::new("sid", &file, "world")) + .await + .expect("write should succeed after permits released"); + let content = fs::read_to_string(&file).expect("read written file"); + assert_eq!(content, "world"); + + let _ = fs::remove_dir_all(workspace); + } }