From 5f655e8f8b79c56d43dcc052078a3f96dc79bdcb Mon Sep 17 00:00:00 2001 From: Adam Perry Date: Fri, 5 Jun 2026 06:07:48 +0000 Subject: [PATCH] fix(rmcp): stabilize OAuth refresh coordination --- codex-rs/Cargo.lock | 2 + codex-rs/rmcp-client/Cargo.toml | 9 + codex-rs/rmcp-client/src/oauth.rs | 270 +++++++++++++++++- codex-rs/rmcp-client/src/rmcp_client.rs | 35 ++- .../streamable_http_oauth_refresh_race.rs | 217 ++++++++++++++ 5 files changed, 507 insertions(+), 26 deletions(-) diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 6da7688435..3226b4d700 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -3579,6 +3579,7 @@ dependencies = [ "codex-utils-pty", "futures", "keyring", + "libc", "memchr", "oauth2", "pretty_assertions", @@ -3597,6 +3598,7 @@ dependencies = [ "urlencoding", "webbrowser", "which 8.0.0", + "windows-sys 0.52.0", "wiremock", ] diff --git a/codex-rs/rmcp-client/Cargo.toml b/codex-rs/rmcp-client/Cargo.toml index 1006ad74e8..3f177975ae 100644 --- a/codex-rs/rmcp-client/Cargo.toml +++ b/codex-rs/rmcp-client/Cargo.toml @@ -74,11 +74,20 @@ wiremock = { workspace = true } [target.'cfg(target_os = "linux")'.dependencies] keyring = { workspace = true, features = ["linux-native-async-persistent"] } +[target.'cfg(unix)'.dependencies] +libc = { workspace = true } + [target.'cfg(target_os = "macos")'.dependencies] keyring = { workspace = true, features = ["apple-native"] } [target.'cfg(target_os = "windows")'.dependencies] keyring = { workspace = true, features = ["windows-native"] } +windows-sys = { version = "0.52", features = [ + "Win32_Foundation", + "Win32_Security", + "Win32_System_SystemInformation", + "Win32_System_Threading", +] } [target.'cfg(any(target_os = "freebsd", target_os = "openbsd"))'.dependencies] keyring = { workspace = true, features = ["sync-secret-service"] } diff --git a/codex-rs/rmcp-client/src/oauth.rs b/codex-rs/rmcp-client/src/oauth.rs index 4816cca315..01799268e7 100644 --- a/codex-rs/rmcp-client/src/oauth.rs +++ b/codex-rs/rmcp-client/src/oauth.rs @@ -531,11 +531,140 @@ fn file_oauth_refresh_lock_path(server_name: &str, url: &str) -> Result fn keyring_oauth_refresh_lock_path(server_name: &str, url: &str) -> Result { let lock_id = oauth_refresh_lock_id(server_name, url)?; - Ok(std::env::temp_dir().join(format!( - "{KEYRING_OAUTH_REFRESH_LOCK_PREFIX}-{lock_id}.lock" + let user_namespace = os_user_namespace()?; + Ok(os_shared_temp_dir()?.join(format!( + "{KEYRING_OAUTH_REFRESH_LOCK_PREFIX}-{user_namespace}-{lock_id}.lock" ))) } +#[cfg(unix)] +fn os_shared_temp_dir() -> Result { + Ok(PathBuf::from("/tmp")) +} + +#[cfg(unix)] +fn os_user_namespace() -> Result { + // SAFETY: getuid has no preconditions and returns the real UID of this process. + Ok(format!("uid-{}", unsafe { libc::getuid() })) +} + +#[cfg(windows)] +fn os_shared_temp_dir() -> Result { + use std::ffi::OsString; + use std::io; + use std::os::windows::ffi::OsStringExt; + use windows_sys::Win32::System::SystemInformation::GetSystemWindowsDirectoryW; + + let mut buffer = vec![0_u16; 32_768]; + // SAFETY: buffer is writable for the length passed to GetSystemWindowsDirectoryW. + let length = unsafe { + GetSystemWindowsDirectoryW( + buffer.as_mut_ptr(), + u32::try_from(buffer.len()).expect("Windows path buffer length fits in u32"), + ) + }; + if length == 0 { + return Err(io::Error::last_os_error()) + .context("failed to resolve the system Windows directory"); + } + let length = usize::try_from(length).context("Windows directory length did not fit usize")?; + if length >= buffer.len() { + return Err(anyhow!( + "system Windows directory exceeded the fixed path buffer" + )); + } + + Ok(PathBuf::from(OsString::from_wide(&buffer[..length])).join("Temp")) +} + +#[cfg(windows)] +fn os_user_namespace() -> Result { + use std::io; + use windows_sys::Win32::Foundation::CloseHandle; + use windows_sys::Win32::Foundation::HANDLE; + use windows_sys::Win32::Security::GetLengthSid; + use windows_sys::Win32::Security::GetTokenInformation; + use windows_sys::Win32::Security::TOKEN_QUERY; + use windows_sys::Win32::Security::TOKEN_USER; + use windows_sys::Win32::Security::TokenUser; + use windows_sys::Win32::System::Threading::GetCurrentProcess; + use windows_sys::Win32::System::Threading::OpenProcessToken; + + struct OwnedHandle(HANDLE); + + impl Drop for OwnedHandle { + fn drop(&mut self) { + if self.0 != 0 { + // SAFETY: the handle was returned by OpenProcessToken and is owned here. + unsafe { + CloseHandle(self.0); + } + } + } + } + + let mut token = 0; + // SAFETY: token points to writable storage for the returned process-token handle. + if unsafe { OpenProcessToken(GetCurrentProcess(), TOKEN_QUERY, &mut token) } == 0 { + return Err(io::Error::last_os_error()).context("failed to open the current process token"); + } + let token = OwnedHandle(token); + + let mut required = 0; + // SAFETY: the null-buffer call obtains the required TOKEN_USER buffer length. + unsafe { + GetTokenInformation(token.0, TokenUser, std::ptr::null_mut(), 0, &mut required); + } + if required == 0 { + return Err(io::Error::last_os_error()) + .context("failed to size the current process user SID"); + } + + let mut buffer = vec![0_u8; required as usize]; + // SAFETY: buffer is writable for required bytes and receives a TOKEN_USER value. + if unsafe { + GetTokenInformation( + token.0, + TokenUser, + buffer.as_mut_ptr().cast(), + required, + &mut required, + ) + } == 0 + { + return Err(io::Error::last_os_error()) + .context("failed to read the current process user SID"); + } + + // SAFETY: GetTokenInformation initialized the buffer with TOKEN_USER. The + // unaligned read avoids assuming Vec has TOKEN_USER alignment. + let token_user = unsafe { std::ptr::read_unaligned(buffer.as_ptr().cast::()) }; + // SAFETY: token_user.User.Sid points into buffer and remains valid here. + let sid_length = unsafe { GetLengthSid(token_user.User.Sid) }; + if sid_length == 0 { + return Err(io::Error::last_os_error()).context("failed to size the current user SID"); + } + // SAFETY: GetLengthSid returned the byte length of the valid SID in buffer. + let sid = unsafe { + std::slice::from_raw_parts(token_user.User.Sid.cast::(), sid_length as usize) + }; + Ok(format!("sid-{}", sha_256_bytes_prefix(sid))) +} + +#[cfg(not(any(unix, windows)))] +fn os_shared_temp_dir() -> Result { + Err(anyhow!( + "keyring OAuth refresh locking is unsupported on this platform" + )) +} + +#[cfg(not(any(unix, windows)))] +fn os_user_namespace() -> Result { + Err(anyhow!( + "keyring OAuth refresh locking is unsupported on this platform" + )) +} + async fn acquire_oauth_refresh_lock(path: &Path) -> Result { let file = open_oauth_lock_file(path)?; loop { @@ -844,12 +973,16 @@ fn write_fallback_file(store: &FallbackFile) -> Result<()> { fn sha_256_prefix(value: &Value) -> Result { let serialized = serde_json::to_string(&value).context("failed to serialize MCP OAuth key payload")?; + Ok(sha_256_bytes_prefix(serialized.as_bytes())) +} + +fn sha_256_bytes_prefix(bytes: &[u8]) -> String { let mut hasher = Sha256::new(); - hasher.update(serialized.as_bytes()); + hasher.update(bytes); let digest = hasher.finalize(); let hex = format!("{digest:x}"); let truncated = &hex[..16]; - Ok(truncated.to_string()) + truncated.to_string() } #[cfg(test)] @@ -866,6 +999,8 @@ mod tests { use codex_keyring_store::tests::MockKeyringStore; + static CODEX_HOME_LOCK: OnceLock> = OnceLock::new(); + struct TempCodexHome { _guard: MutexGuard<'static, ()>, _dir: tempfile::TempDir, @@ -873,8 +1008,7 @@ mod tests { impl TempCodexHome { fn new() -> Self { - static LOCK: OnceLock> = OnceLock::new(); - let guard = LOCK + let guard = CODEX_HOME_LOCK .get_or_init(Mutex::default) .lock() .unwrap_or_else(PoisonError::into_inner); @@ -887,6 +1021,23 @@ mod tests { _dir: dir, } } + + fn unusable() -> Self { + let guard = CODEX_HOME_LOCK + .get_or_init(Mutex::default) + .lock() + .unwrap_or_else(PoisonError::into_inner); + let dir = tempdir().expect("create CODEX_HOME parent temp dir"); + let file = dir.path().join("not-a-directory"); + fs::write(&file, "not a directory").expect("create unusable CODEX_HOME path"); + unsafe { + std::env::set_var("CODEX_HOME", file); + } + Self { + _guard: guard, + _dir: dir, + } + } } impl Drop for TempCodexHome { @@ -904,7 +1055,16 @@ mod tests { let other_url = "https://other.example.test/mcp"; let keyring_path = keyring_oauth_refresh_lock_path(&tokens.server_name, &tokens.url)?; - assert_eq!(keyring_path.parent(), Some(std::env::temp_dir().as_path())); + let shared_temp_dir = os_shared_temp_dir()?; + let user_namespace = os_user_namespace()?; + assert_eq!(keyring_path.parent(), Some(shared_temp_dir.as_path())); + assert!( + keyring_path + .file_name() + .is_some_and(|name| name.to_string_lossy().contains(&user_namespace)) + ); + #[cfg(unix)] + assert_eq!(shared_temp_dir, PathBuf::from("/tmp")); assert_eq!( keyring_path, keyring_oauth_refresh_lock_path(&tokens.server_name, &tokens.url)? @@ -920,6 +1080,102 @@ mod tests { Ok(()) } + #[tokio::test(flavor = "current_thread")] + async fn keyring_refresh_locking_works_without_codex_home_for_auto_and_keyring() -> Result<()> { + let _env = TempCodexHome::unusable(); + let store = MockKeyringStore::default(); + let tokens = sample_tokens(); + let serialized = serde_json::to_string(&tokens)?; + let key = compute_store_key(&tokens.server_name, &tokens.url)?; + store.save(KEYRING_SERVICE, &key, &serialized)?; + + let held_lock = acquire_oauth_refresh_lock(&keyring_oauth_refresh_lock_path( + &tokens.server_name, + &tokens.url, + )?) + .await?; + let blocked = tokio::time::timeout( + Duration::from_millis(50), + lock_and_load_oauth_tokens_with_keyring( + &store, + &tokens.server_name, + &tokens.url, + OAuthCredentialsStoreMode::Keyring, + ), + ) + .await; + assert!( + blocked.is_err(), + "same keyring credential should wait for its refresh lock" + ); + drop(held_lock); + + for store_mode in [ + OAuthCredentialsStoreMode::Keyring, + OAuthCredentialsStoreMode::Auto, + ] { + let (locks, loaded) = lock_and_load_oauth_tokens_with_keyring( + &store, + &tokens.server_name, + &tokens.url, + store_mode, + ) + .await?; + assert_eq!(locks.len(), 1); + assert_tokens_match_without_expiry( + &loaded.expect("keyring credentials should load"), + &tokens, + ); + } + Ok(()) + } + + #[tokio::test(flavor = "current_thread")] + async fn distinct_keyring_credentials_do_not_block_each_other() -> Result<()> { + let _env = TempCodexHome::unusable(); + let store = MockKeyringStore::default(); + let first = sample_tokens(); + let mut second = sample_tokens(); + second.url = "https://other.example.test/mcp".to_string(); + for tokens in [&first, &second] { + let key = compute_store_key(&tokens.server_name, &tokens.url)?; + store.save(KEYRING_SERVICE, &key, &serde_json::to_string(tokens)?)?; + } + + let held_lock = acquire_oauth_refresh_lock(&keyring_oauth_refresh_lock_path( + &first.server_name, + &first.url, + )?) + .await?; + for store_mode in [ + OAuthCredentialsStoreMode::Keyring, + OAuthCredentialsStoreMode::Auto, + ] { + let result = tokio::time::timeout( + Duration::from_millis(500), + lock_and_load_oauth_tokens_with_keyring( + &store, + &second.server_name, + &second.url, + store_mode, + ), + ) + .await; + assert!( + result.is_ok(), + "a distinct keyring credential should not wait for another lock" + ); + let (locks, loaded) = result.expect("timeout checked above")?; + assert_eq!(locks.len(), 1); + assert_tokens_match_without_expiry( + &loaded.expect("second keyring credential should load"), + &second, + ); + } + drop(held_lock); + Ok(()) + } + #[tokio::test(flavor = "current_thread")] async fn refresh_lock_loading_covers_keyring_and_auto_branches() -> Result<()> { let _env = TempCodexHome::new(); diff --git a/codex-rs/rmcp-client/src/rmcp_client.rs b/codex-rs/rmcp-client/src/rmcp_client.rs index 2396b21248..6bc252afae 100644 --- a/codex-rs/rmcp-client/src/rmcp_client.rs +++ b/codex-rs/rmcp-client/src/rmcp_client.rs @@ -440,7 +440,6 @@ impl RmcpClient { params: Option, timeout: Option, ) -> Result { - self.refresh_oauth_if_needed().await; let result = self .run_service_operation("tools/list", timeout, move |service| { let params = params.clone(); @@ -456,7 +455,6 @@ impl RmcpClient { params: Option, timeout: Option, ) -> Result { - self.refresh_oauth_if_needed().await; let result = self .run_service_operation("tools/list", timeout, move |service| { let params = params.clone(); @@ -501,7 +499,6 @@ impl RmcpClient { params: Option, timeout: Option, ) -> Result { - self.refresh_oauth_if_needed().await; let result = self .run_service_operation("resources/list", timeout, move |service| { let params = params.clone(); @@ -517,7 +514,6 @@ impl RmcpClient { params: Option, timeout: Option, ) -> Result { - self.refresh_oauth_if_needed().await; let result = self .run_service_operation("resources/templates/list", timeout, move |service| { let params = params.clone(); @@ -533,7 +529,6 @@ impl RmcpClient { params: ReadResourceRequestParams, timeout: Option, ) -> Result { - self.refresh_oauth_if_needed().await; let result = self .run_service_operation("resources/read", timeout, move |service| { let params = params.clone(); @@ -551,7 +546,6 @@ impl RmcpClient { meta: Option, timeout: Option, ) -> Result { - self.refresh_oauth_if_needed().await; let arguments = match arguments { Some(Value::Object(map)) => Some(map), Some(other) => { @@ -607,7 +601,6 @@ impl RmcpClient { method: &str, params: Option, ) -> Result<()> { - self.refresh_oauth_if_needed().await; self.run_service_operation( "notifications/custom", /*timeout*/ None, @@ -637,7 +630,6 @@ impl RmcpClient { method: &str, params: Option, ) -> Result { - self.refresh_oauth_if_needed().await; let response = self .run_service_operation("requests/custom", /*timeout*/ None, move |service| { let params = params.clone(); @@ -701,14 +693,6 @@ impl RmcpClient { } } - async fn refresh_oauth_if_needed(&self) { - if let Some(runtime) = self.oauth_persistor().await - && let Err(error) = runtime.refresh_if_needed().await - { - warn!("failed to refresh OAuth tokens: {error}"); - } - } - async fn create_pending_transport( transport_recipe: &TransportRecipe, ) -> Result { @@ -876,8 +860,10 @@ impl RmcpClient { Fut: std::future::Future>, { let service = self.service().await?; + let oauth_persistor = self.oauth_persistor().await; match Self::run_service_operation_once( Arc::clone(&service), + oauth_persistor, label, timeout, self.elicitation_pause_state.clone(), @@ -889,8 +875,10 @@ impl RmcpClient { Err(error) if Self::is_session_expired_404(&error) => { self.reinitialize_after_session_expiry(&service).await?; let recovered_service = self.service().await?; + let oauth_persistor = self.oauth_persistor().await; Self::run_service_operation_once( recovered_service, + oauth_persistor, label, timeout, self.elicitation_pause_state.clone(), @@ -905,6 +893,7 @@ impl RmcpClient { async fn run_service_operation_once( service: Arc>, + oauth_persistor: Option, label: &str, timeout: Option, pause_state: ElicitationPauseState, @@ -914,17 +903,25 @@ impl RmcpClient { F: Fn(Arc>) -> Fut, Fut: std::future::Future>, { + let refresh_and_operation = async move { + if let Some(runtime) = oauth_persistor + && let Err(error) = runtime.refresh_if_needed().await + { + warn!("failed to refresh OAuth tokens: {error}"); + } + operation(service).await.map_err(ClientOperationError::from) + }; + match timeout { Some(duration) => { - active_time_timeout(duration, pause_state.subscribe(), operation(service)) + active_time_timeout(duration, pause_state.subscribe(), refresh_and_operation) .await .map_err(|_| ClientOperationError::Timeout { label: label.to_string(), duration, })? - .map_err(ClientOperationError::from) } - None => operation(service).await.map_err(ClientOperationError::from), + None => refresh_and_operation.await, } } diff --git a/codex-rs/rmcp-client/tests/streamable_http_oauth_refresh_race.rs b/codex-rs/rmcp-client/tests/streamable_http_oauth_refresh_race.rs index c2387ec10f..78ef91686e 100644 --- a/codex-rs/rmcp-client/tests/streamable_http_oauth_refresh_race.rs +++ b/codex-rs/rmcp-client/tests/streamable_http_oauth_refresh_race.rs @@ -52,6 +52,7 @@ const NEW_ACCESS_TOKEN: &str = "new-access-token"; const NEW_REFRESH_TOKEN: &str = "new-one-time-refresh-token"; const KEYRING_SERVICE: &str = "Codex MCP Credentials"; const CHILD_WAIT_TIMEOUT: Duration = Duration::from_secs(10); +const OPERATION_TIMEOUT: Duration = Duration::from_millis(100); const SERVER_URL_ENV: &str = "MCP_TEST_OAUTH_RACE_SERVER_URL"; const READY_PATH_ENV: &str = "MCP_TEST_OAUTH_RACE_READY_PATH"; @@ -225,6 +226,168 @@ async fn refresh_lock_wait_uses_startup_timeout_budget() -> anyhow::Result<()> { Ok(()) } +#[tokio::test(flavor = "multi_thread", worker_threads = 1)] +async fn refresh_lock_wait_uses_operation_timeout_budget() -> anyhow::Result<()> { + let server = MockServer::start().await; + mount_oauth_metadata(&server, /*expected_requests*/ 1).await; + Mock::given(method("POST")) + .and(path("/oauth/token")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "access_token": NEW_ACCESS_TOKEN, + "token_type": "Bearer", + "expires_in": 30, + "refresh_token": NEW_REFRESH_TOKEN, + }))) + .expect(1) + .mount(&server) + .await; + mount_mcp_server(&server, /*expected_requests*/ 2).await; + + let codex_home = TempDir::new()?; + let control_dir = TempDir::new()?; + let server_url = format!("{}/mcp", server.uri()); + seed_tokens(&codex_home, &server_url, /*expires_at*/ 0).await?; + + let ready_path = control_dir.path().join("client.ready"); + let go_path = control_dir.path().join("client.go"); + let result_path = control_dir.path().join("client.result"); + let child = spawn_operation_timeout_child( + &codex_home, + &server_url, + &ready_path, + &go_path, + &result_path, + )?; + wait_for_paths(std::slice::from_ref(&ready_path)).await?; + + let lock_path = file_oauth_refresh_lock_path(&codex_home, &server_url)?; + let refresh_lock = OpenOptions::new() + .read(true) + .write(true) + .create(true) + .truncate(false) + .open(lock_path)?; + refresh_lock.lock()?; + + fs::write(&go_path, "go")?; + wait_for_children(vec![child]).await?; + assert_eq!( + fs::read_to_string(result_path)?, + format!("error:timed out awaiting tools/list after {OPERATION_TIMEOUT:?}") + ); + + drop(refresh_lock); + server.verify().await; + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 1)] +async fn refresh_provider_call_uses_operation_timeout_budget() -> anyhow::Result<()> { + let server = MockServer::start().await; + mount_oauth_metadata(&server, /*expected_requests*/ 1).await; + let refresh_attempts = Arc::new(AtomicUsize::new(0)); + let responder_attempts = Arc::clone(&refresh_attempts); + Mock::given(method("POST")) + .and(path("/oauth/token")) + .respond_with(move |_request: &Request| { + let response = ResponseTemplate::new(200).set_body_json(json!({ + "access_token": NEW_ACCESS_TOKEN, + "token_type": "Bearer", + "expires_in": 30, + "refresh_token": NEW_REFRESH_TOKEN, + })); + if responder_attempts.fetch_add(1, Ordering::SeqCst) == 0 { + response + } else { + response.set_delay(Duration::from_secs(1)) + } + }) + .expect(2) + .mount(&server) + .await; + mount_mcp_server(&server, /*expected_requests*/ 2).await; + + let codex_home = TempDir::new()?; + let control_dir = TempDir::new()?; + let server_url = format!("{}/mcp", server.uri()); + seed_tokens(&codex_home, &server_url, /*expires_at*/ 0).await?; + + let ready_path = control_dir.path().join("client.ready"); + let go_path = control_dir.path().join("client.go"); + let result_path = control_dir.path().join("client.result"); + let child = spawn_operation_timeout_child( + &codex_home, + &server_url, + &ready_path, + &go_path, + &result_path, + )?; + wait_for_paths(std::slice::from_ref(&ready_path)).await?; + fs::write(&go_path, "go")?; + wait_for_children(vec![child]).await?; + + assert_eq!( + fs::read_to_string(result_path)?, + format!("error:timed out awaiting tools/list after {OPERATION_TIMEOUT:?}") + ); + assert_eq!(refresh_attempts.load(Ordering::SeqCst), 2); + + server.verify().await; + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 1)] +async fn distinct_file_credentials_do_not_block_each_other() -> anyhow::Result<()> { + let server = MockServer::start().await; + mount_oauth_metadata(&server, /*expected_requests*/ 1).await; + Mock::given(method("POST")) + .and(path("/oauth/token")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "access_token": NEW_ACCESS_TOKEN, + "token_type": "Bearer", + "expires_in": 7200, + "refresh_token": NEW_REFRESH_TOKEN, + }))) + .expect(1) + .mount(&server) + .await; + mount_mcp_server(&server, /*expected_requests*/ 2).await; + + let codex_home = TempDir::new()?; + let control_dir = TempDir::new()?; + let server_url = format!("{}/mcp", server.uri()); + seed_tokens(&codex_home, &server_url, /*expires_at*/ 0).await?; + + let unrelated_lock_path = + file_oauth_refresh_lock_path(&codex_home, "https://unrelated.example.test/mcp")?; + let unrelated_lock = OpenOptions::new() + .read(true) + .write(true) + .create(true) + .truncate(false) + .open(unrelated_lock_path)?; + unrelated_lock.lock()?; + + let ready_path = control_dir.path().join("client.ready"); + let go_path = control_dir.path().join("client.go"); + let result_path = control_dir.path().join("client.result"); + let child = spawn_client_child( + &codex_home, + &server_url, + &ready_path, + &go_path, + &result_path, + )?; + wait_for_paths(std::slice::from_ref(&ready_path)).await?; + fs::write(&go_path, "go")?; + wait_for_children(vec![child]).await?; + + assert_eq!(fs::read_to_string(result_path)?, "success"); + drop(unrelated_lock); + server.verify().await; + Ok(()) +} + #[tokio::test(flavor = "multi_thread", worker_threads = 1)] async fn rejected_refresh_requires_login_without_reaching_mcp() -> anyhow::Result<()> { let server = MockServer::start().await; @@ -433,6 +596,36 @@ async fn oauth_refresh_race_client_child() -> anyhow::Result<()> { Ok(()) } +#[tokio::test(flavor = "multi_thread", worker_threads = 1)] +#[ignore = "spawned by the OAuth refresh race integration tests"] +async fn oauth_refresh_operation_timeout_child() -> anyhow::Result<()> { + let server_url = std::env::var(SERVER_URL_ENV)?; + let ready_path = PathBuf::from(std::env::var(READY_PATH_ENV)?); + let go_path = PathBuf::from(std::env::var(GO_PATH_ENV)?); + let result_path = PathBuf::from(std::env::var(RESULT_PATH_ENV)?); + let client = RmcpClient::new_streamable_http_client( + SERVER_NAME, + &server_url, + /*bearer_token*/ None, + /*http_headers*/ None, + /*env_http_headers*/ None, + OAuthCredentialsStoreMode::File, + Environment::default_for_tests().get_http_client(), + /*auth_provider*/ None, + ) + .await?; + initialize_client(&client).await?; + + fs::write(ready_path, "ready")?; + wait_for_paths(std::slice::from_ref(&go_path)).await?; + let outcome = match client.list_tools(None, Some(OPERATION_TIMEOUT)).await { + Ok(_) => "success".to_string(), + Err(error) => format!("error:{error:#}"), + }; + fs::write(result_path, outcome)?; + Ok(()) +} + async fn mount_oauth_metadata(server: &MockServer, expected_requests: u64) { Mock::given(method("GET")) .and(path("/.well-known/oauth-authorization-server/mcp")) @@ -529,6 +722,30 @@ fn spawn_client_child( Ok(command.spawn()?) } +fn spawn_operation_timeout_child( + codex_home: &TempDir, + server_url: &str, + ready_path: &Path, + go_path: &Path, + result_path: &Path, +) -> anyhow::Result { + let mut command = Command::new(std::env::current_exe()?); + command + .args([ + "oauth_refresh_operation_timeout_child", + "--exact", + "--ignored", + "--nocapture", + ]) + .env("CODEX_HOME", codex_home.path()) + .env(SERVER_URL_ENV, server_url) + .env(READY_PATH_ENV, ready_path) + .env(GO_PATH_ENV, go_path) + .env(RESULT_PATH_ENV, result_path) + .kill_on_drop(true); + Ok(command.spawn()?) +} + async fn wait_for_children(children: Vec) -> anyhow::Result<()> { for mut child in children { let status = timeout(CHILD_WAIT_TIMEOUT, child.wait())