fix(rmcp): stabilize OAuth refresh coordination

2026-10-26 18:38:47 +00:00 · 2026-06-05 06:07:48 +00:00
parent 0d63af1ab7
commit 5f655e8f8b
5 changed files with 507 additions and 26 deletions
@@ -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",
 ]

@@ -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"] }
@@ -531,11 +531,140 @@ fn file_oauth_refresh_lock_path(server_name: &str, url: &str) -> Result<PathBuf>

 fn keyring_oauth_refresh_lock_path(server_name: &str, url: &str) -> Result<PathBuf> {
    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<PathBuf> {
+    Ok(PathBuf::from("/tmp"))
+}
+
+#[cfg(unix)]
+fn os_user_namespace() -> Result<String> {
+    // 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<PathBuf> {
+    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<String> {
+    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<u8> has TOKEN_USER alignment.
+    let token_user = unsafe { std::ptr::read_unaligned(buffer.as_ptr().cast::<TOKEN_USER>()) };
+    // 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::<u8>(), sid_length as usize)
+    };
+    Ok(format!("sid-{}", sha_256_bytes_prefix(sid)))
+}
+
+#[cfg(not(any(unix, windows)))]
+fn os_shared_temp_dir() -> Result<PathBuf> {
+    Err(anyhow!(
+        "keyring OAuth refresh locking is unsupported on this platform"
+    ))
+}
+
+#[cfg(not(any(unix, windows)))]
+fn os_user_namespace() -> Result<String> {
+    Err(anyhow!(
+        "keyring OAuth refresh locking is unsupported on this platform"
+    ))
+}
+
 async fn acquire_oauth_refresh_lock(path: &Path) -> Result<fs::File> {
    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<String> {
    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<Mutex<()>> = OnceLock::new();
+
    struct TempCodexHome {
        _guard: MutexGuard<'static, ()>,
        _dir: tempfile::TempDir,
@@ -873,8 +1008,7 @@ mod tests {

    impl TempCodexHome {
        fn new() -> Self {
-            static LOCK: OnceLock<Mutex<()>> = 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();
@@ -440,7 +440,6 @@ impl RmcpClient {
        params: Option<PaginatedRequestParams>,
        timeout: Option<Duration>,
    ) -> Result<ListToolsResult> {
-        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<PaginatedRequestParams>,
        timeout: Option<Duration>,
    ) -> Result<ListToolsWithConnectorIdResult> {
-        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<PaginatedRequestParams>,
        timeout: Option<Duration>,
    ) -> Result<ListResourcesResult> {
-        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<PaginatedRequestParams>,
        timeout: Option<Duration>,
    ) -> Result<ListResourceTemplatesResult> {
-        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<Duration>,
    ) -> Result<ReadResourceResult> {
-        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<serde_json::Value>,
        timeout: Option<Duration>,
    ) -> Result<CallToolResult> {
-        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<serde_json::Value>,
    ) -> 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<serde_json::Value>,
    ) -> Result<ServerResult> {
-        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<PendingTransport> {
@@ -876,8 +860,10 @@ impl RmcpClient {
        Fut: std::future::Future<Output = std::result::Result<T, rmcp::service::ServiceError>>,
    {
        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<T, F, Fut>(
        service: Arc<RunningService<RoleClient, ElicitationClientService>>,
+        oauth_persistor: Option<OAuthPersistor>,
        label: &str,
        timeout: Option<Duration>,
        pause_state: ElicitationPauseState,
@@ -914,17 +903,25 @@ impl RmcpClient {
        F: Fn(Arc<RunningService<RoleClient, ElicitationClientService>>) -> Fut,
        Fut: std::future::Future<Output = std::result::Result<T, rmcp::service::ServiceError>>,
    {
+        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,
        }
    }

@@ -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<Child> {
+    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<Child>) -> anyhow::Result<()> {
    for mut child in children {
        let status = timeout(CHILD_WAIT_TIMEOUT, child.wait())