fix(rmcp-client): harden oauth refresh persistence

codex-rs/rmcp-client/src/bin/test_streamable_http_server.rs

@@ -4,6 +4,8 @@ use std::fs;
 use std::io::ErrorKind;
 use std::net::SocketAddr;
 use std::sync::Arc;
+use std::sync::atomic::AtomicUsize;
+use std::sync::atomic::Ordering;
 use std::time::Duration;
 
 use axum::Router;
@@ -55,6 +57,8 @@ use tokio::task;
 use tokio::time::sleep;
 use urlencoding::decode;
 
+static REFRESH_TOKEN_USES: AtomicUsize = AtomicUsize::new(0);
+
 #[derive(Clone)]
 struct TestToolServer {
     tools: Arc<Vec<Tool>>,
@@ -168,6 +172,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
             session_failure_state.clone(),
             fail_session_post_when_armed,
         ))
+        .layer(middleware::from_fn(delay_mcp_initialize_when_configured))
         .with_state(session_failure_state);
 
     let router = if let Ok(token) = std::env::var("MCP_EXPECT_BEARER") {
@@ -413,18 +418,10 @@ async fn exchange_refresh_token(request: Request<Body>) -> Result<Response, Stat
         .map_err(|_| StatusCode::BAD_REQUEST)?;
     let params = parse_form_body(&body)?;
 
-    if let Ok(delay_ms) = std::env::var("MCP_REFRESH_TOKEN_DELAY_MS")
-        && let Ok(delay_ms) = delay_ms.parse::<u64>()
-    {
-        sleep(Duration::from_millis(delay_ms)).await;
-    }
-
     let expected_refresh_token =
         std::env::var("MCP_EXPECT_REFRESH_TOKEN").map_err(|_| StatusCode::BAD_REQUEST)?;
     let access_token =
         std::env::var("MCP_REFRESH_ACCESS_TOKEN").map_err(|_| StatusCode::BAD_REQUEST)?;
-    let refresh_token =
-        std::env::var("MCP_ROTATED_REFRESH_TOKEN").map_err(|_| StatusCode::BAD_REQUEST)?;
 
     if params.get("grant_type").map(String::as_str) != Some("refresh_token")
         || params.get("refresh_token").map(String::as_str) != Some(expected_refresh_token.as_str())
@@ -432,22 +429,55 @@ async fn exchange_refresh_token(request: Request<Body>) -> Result<Response, Stat
         return Err(StatusCode::UNAUTHORIZED);
     }
 
+    if let Ok(max_uses) = std::env::var("MCP_REFRESH_TOKEN_MAX_USES")
+        && let Ok(max_uses) = max_uses.parse::<usize>()
+    {
+        let use_count = REFRESH_TOKEN_USES.fetch_add(1, Ordering::SeqCst) + 1;
+        if use_count > max_uses {
+            return Err(StatusCode::UNAUTHORIZED);
+        }
+    }
+
+    if let Ok(delay_ms) = std::env::var("MCP_REFRESH_TOKEN_DELAY_MS")
+        && let Ok(delay_ms) = delay_ms.parse::<u64>()
+    {
+        sleep(Duration::from_millis(delay_ms)).await;
+    }
+
+    let mut token_response = json!({
+        "access_token": access_token,
+        "token_type": "Bearer",
+        "expires_in": 7200,
+    });
+    if std::env::var("MCP_OMIT_ROTATED_REFRESH_TOKEN").is_err() {
+        let refresh_token =
+            std::env::var("MCP_ROTATED_REFRESH_TOKEN").map_err(|_| StatusCode::BAD_REQUEST)?;
+        token_response["refresh_token"] = json!(refresh_token);
+    }
+
     #[expect(clippy::expect_used)]
     Ok(Response::builder()
         .status(StatusCode::OK)
         .header(CONTENT_TYPE, "application/json")
         .body(Body::from(
-            serde_json::to_vec(&json!({
-                "access_token": access_token,
-                "token_type": "Bearer",
-                "expires_in": 7200,
-                "refresh_token": refresh_token,
-            }))
-            .expect("failed to serialize token response"),
+            serde_json::to_vec(&token_response).expect("failed to serialize token response"),
         ))
         .expect("valid token response"))
 }
 
+async fn delay_mcp_initialize_when_configured(request: Request<Body>, next: Next) -> Response {
+    if request.uri().path() == "/mcp"
+        && request.method() == Method::POST
+        && !request.headers().contains_key(MCP_SESSION_ID_HEADER)
+        && let Ok(delay_ms) = std::env::var("MCP_INITIALIZE_DELAY_MS")
+        && let Ok(delay_ms) = delay_ms.parse::<u64>()
+    {
+        sleep(Duration::from_millis(delay_ms)).await;
+    }
+
+    next.run(request).await
+}
+
 fn parse_form_body(body: &[u8]) -> Result<HashMap<String, String>, StatusCode> {
     let body = std::str::from_utf8(body).map_err(|_| StatusCode::BAD_REQUEST)?;
     body.split('&')

codex-rs/rmcp-client/src/oauth.rs

@@ -35,9 +35,11 @@ use sha2::Digest;
 use sha2::Sha256;
 use std::collections::BTreeMap;
 use std::fs;
+use std::fs::OpenOptions;
 use std::io::ErrorKind;
 use std::path::PathBuf;
 use std::sync::Arc;
+use std::sync::OnceLock;
 use std::time::Duration;
 use std::time::SystemTime;
 use std::time::UNIX_EPOCH;
@@ -45,13 +47,19 @@ use tracing::warn;
 
 use codex_keyring_store::DefaultKeyringStore;
 use codex_keyring_store::KeyringStore;
+use rmcp::transport::auth::AuthError;
 use rmcp::transport::auth::AuthorizationManager;
+use rmcp::transport::auth::CredentialStore;
+use rmcp::transport::auth::InMemoryCredentialStore;
+use rmcp::transport::auth::StoredCredentials;
 use tokio::sync::Mutex;
 
 use codex_utils_home_dir::find_codex_home;
 
 const KEYRING_SERVICE: &str = "Codex MCP Credentials";
 const REFRESH_SKEW_MILLIS: u64 = 30_000;
+const REFRESH_LOCK_RETRIES: usize = 200;
+const REFRESH_LOCK_RETRY_SLEEP: Duration = Duration::from_millis(25);
 
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
 pub struct StoredOAuthTokens {
@@ -259,7 +267,8 @@ struct OAuthPersistorInner {
     url: String,
     authorization_manager: Arc<Mutex<AuthorizationManager>>,
     store_mode: OAuthCredentialsStoreMode,
-    last_credentials: Mutex<Option<StoredOAuthTokens>>,
+    current_credentials: Mutex<Option<StoredOAuthTokens>>,
+    persisted_credentials: Mutex<Option<StoredOAuthTokens>>,
 }
 
 impl OAuthPersistor {
@@ -276,7 +285,8 @@ impl OAuthPersistor {
                 url,
                 authorization_manager,
                 store_mode,
-                last_credentials: Mutex::new(initial_credentials),
+                current_credentials: Mutex::new(initial_credentials.clone()),
+                persisted_credentials: Mutex::new(initial_credentials),
             }),
         }
     }
@@ -295,15 +305,28 @@ impl OAuthPersistor {
         }?;
 
         match maybe_credentials {
-            Some(credentials) => {
-                let mut last_credentials = self.inner.last_credentials.lock().await;
+            Some(mut credentials) => {
+                let current_credentials = self.inner.current_credentials.lock().await.clone();
+                if credentials.refresh_token().is_none()
+                    && let Some(refresh_token) = current_credentials
+                        .as_ref()
+                        .and_then(|prev| prev.token_response.0.refresh_token())
+                {
+                    credentials
+                        .set_refresh_token(Some(RefreshToken::new(refresh_token.secret().clone())));
+                    self.replace_manager_credentials(&client_id, credentials.clone())
+                        .await?;
+                }
+
                 let new_token_response = WrappedOAuthTokenResponse(credentials.clone());
-                let same_token = last_credentials
+                let same_token = current_credentials
                     .as_ref()
                     .map(|prev| prev.token_response == new_token_response)
                     .unwrap_or(false);
                 let expires_at = if same_token {
-                    last_credentials.as_ref().and_then(|prev| prev.expires_at)
+                    current_credentials
+                        .as_ref()
+                        .and_then(|prev| prev.expires_at)
                 } else {
                     compute_expires_at_millis(&credentials)
                 };
@@ -314,14 +337,25 @@ impl OAuthPersistor {
                     token_response: new_token_response,
                     expires_at,
                 };
-                if last_credentials.as_ref() != Some(&stored) {
+                {
+                    let mut current_credentials = self.inner.current_credentials.lock().await;
+                    *current_credentials = Some(stored.clone());
+                }
+
+                let mut persisted_credentials = self.inner.persisted_credentials.lock().await;
+                if persisted_credentials.as_ref() != Some(&stored) {
                     save_oauth_tokens(&self.inner.server_name, &stored, self.inner.store_mode)?;
-                    *last_credentials = Some(stored);
+                    *persisted_credentials = Some(stored);
                 }
             }
             None => {
-                let mut last_serialized = self.inner.last_credentials.lock().await;
-                if last_serialized.take().is_some()
+                {
+                    let mut current_credentials = self.inner.current_credentials.lock().await;
+                    *current_credentials = None;
+                }
+
+                let mut persisted_credentials = self.inner.persisted_credentials.lock().await;
+                if persisted_credentials.take().is_some()
                     && let Err(error) = delete_oauth_tokens(
                         &self.inner.server_name,
                         &self.inner.url,
@@ -344,8 +378,46 @@ impl OAuthPersistor {
         reason = "AuthorizationManager async access must be serialized through its mutex"
     )]
     pub(crate) async fn refresh_if_needed(&self) -> Result<()> {
+        let refresh_lock = refresh_lock_for(&self.inner.server_name, &self.inner.url);
+        let _refresh_guard = refresh_lock.lock_owned().await;
+        let _refresh_file_lock =
+            acquire_refresh_file_lock(&self.inner.server_name, &self.inner.url).await?;
+
+        match load_oauth_tokens(
+            &self.inner.server_name,
+            &self.inner.url,
+            self.inner.store_mode,
+        ) {
+            Ok(Some(tokens)) => {
+                let current_credentials = self.inner.current_credentials.lock().await.clone();
+                let persisted_credentials = self.inner.persisted_credentials.lock().await.clone();
+                if current_credentials.as_ref() != Some(&tokens)
+                    && persisted_credentials.as_ref() != Some(&tokens)
+                {
+                    self.replace_manager_credentials(
+                        &tokens.client_id,
+                        tokens.token_response.0.clone(),
+                    )
+                    .await?;
+                    {
+                        let mut current_credentials = self.inner.current_credentials.lock().await;
+                        *current_credentials = Some(tokens.clone());
+                    }
+                    let mut persisted_credentials = self.inner.persisted_credentials.lock().await;
+                    *persisted_credentials = Some(tokens);
+                }
+            }
+            Ok(None) => {}
+            Err(error) => {
+                warn!(
+                    "failed to reload OAuth tokens for server {} before refresh: {error}",
+                    self.inner.server_name
+                );
+            }
+        }
+
         let expires_at = {
-            let guard = self.inner.last_credentials.lock().await;
+            let guard = self.inner.current_credentials.lock().await;
             guard.as_ref().and_then(|tokens| tokens.expires_at)
         };
 
@@ -353,19 +425,131 @@ impl OAuthPersistor {
             return Ok(());
         }
 
-        {
+        let previous_credentials = self.inner.current_credentials.lock().await.clone();
+        let previous_refresh_token = previous_credentials
+            .as_ref()
+            .and_then(|tokens| tokens.token_response.0.refresh_token())
+            .map(|token| token.secret().clone());
+        let mut refreshed_credentials = {
             let manager = self.inner.authorization_manager.clone();
             let guard = manager.lock().await;
-            guard.refresh_token().await.with_context(|| {
-                format!(
-                    "failed to refresh OAuth tokens for server {}",
-                    self.inner.server_name
-                )
-            })?;
+            match guard.refresh_token().await {
+                Ok(credentials) => credentials,
+                Err(AuthError::AuthorizationRequired | AuthError::TokenRefreshFailed(_)) => {
+                    return Err(anyhow::anyhow!("Auth required for server"));
+                }
+                Err(error) => {
+                    return Err(Error::new(error).context(format!(
+                        "failed to refresh OAuth tokens for server {}",
+                        self.inner.server_name
+                    )));
+                }
+            }
+        };
+
+        if refreshed_credentials.refresh_token().is_none()
+            && let Some(refresh_token) = previous_refresh_token
+            && let Some(previous_credentials) = previous_credentials.as_ref()
+        {
+            refreshed_credentials.set_refresh_token(Some(RefreshToken::new(refresh_token)));
+            self.replace_manager_credentials(
+                &previous_credentials.client_id,
+                refreshed_credentials.clone(),
+            )
+            .await?;
         }
 
-        self.persist_if_needed().await
+        if let Err(error) = self.persist_if_needed().await {
+            warn!(
+                "failed to persist refreshed OAuth tokens for server {}: {error}",
+                self.inner.server_name
+            );
+        }
+
+        Ok(())
     }
+
+    async fn replace_manager_credentials(
+        &self,
+        client_id: &str,
+        token_response: OAuthTokenResponse,
+    ) -> Result<()> {
+        let store = InMemoryCredentialStore::new();
+        store
+            .save(StoredCredentials::new(
+                client_id.to_string(),
+                Some(token_response.clone()),
+                token_response
+                    .scopes()
+                    .map(|scopes| scopes.iter().map(|scope| scope.to_string()).collect())
+                    .unwrap_or_default(),
+                Some(
+                    SystemTime::now()
+                        .duration_since(UNIX_EPOCH)
+                        .unwrap_or_else(|_| Duration::from_secs(0))
+                        .as_secs(),
+                ),
+            ))
+            .await?;
+
+        let manager = self.inner.authorization_manager.clone();
+        let mut guard = manager.lock().await;
+        guard.configure_client_id(client_id)?;
+        guard.set_credential_store(store);
+        Ok(())
+    }
+}
+
+fn refresh_lock_for(server_name: &str, url: &str) -> Arc<Mutex<()>> {
+    static REFRESH_LOCKS: OnceLock<std::sync::Mutex<BTreeMap<String, Arc<Mutex<()>>>>> =
+        OnceLock::new();
+
+    let mut locks = REFRESH_LOCKS
+        .get_or_init(std::sync::Mutex::default)
+        .lock()
+        .unwrap_or_else(std::sync::PoisonError::into_inner);
+    locks
+        .entry(format!("{server_name}\n{url}"))
+        .or_insert_with(|| Arc::new(Mutex::new(())))
+        .clone()
+}
+
+struct RefreshFileLock {
+    _file: fs::File,
+}
+
+async fn acquire_refresh_file_lock(server_name: &str, url: &str) -> Result<RefreshFileLock> {
+    let path = refresh_file_lock_path(server_name, url)?;
+    if let Some(parent) = path.parent() {
+        fs::create_dir_all(parent)?;
+    }
+    let file = OpenOptions::new()
+        .read(true)
+        .write(true)
+        .create(true)
+        .truncate(false)
+        .open(path)?;
+
+    for _ in 0..REFRESH_LOCK_RETRIES {
+        match file.try_lock() {
+            Ok(()) => return Ok(RefreshFileLock { _file: file }),
+            Err(std::fs::TryLockError::WouldBlock) => {
+                tokio::time::sleep(REFRESH_LOCK_RETRY_SLEEP).await;
+            }
+            Err(error) => return Err(error.into()),
+        }
+    }
+
+    Err(anyhow::anyhow!(
+        "timed out waiting for OAuth refresh lock for server {server_name}"
+    ))
+}
+
+fn refresh_file_lock_path(server_name: &str, url: &str) -> Result<PathBuf> {
+    let digest = sha_256_prefix(&Value::String(format!("{server_name}\n{url}")))?;
+    Ok(find_codex_home()?
+        .join(format!(".mcp-oauth-refresh-{digest}.lock"))
+        .to_path_buf())
 }
 
 const FALLBACK_FILENAME: &str = ".credentials.json";

codex-rs/rmcp-client/src/rmcp_client.rs

@@ -843,11 +843,21 @@ impl RmcpClient {
                 oauth_persistor,
             } => {
                 match timeout {
-                    Some(duration) => time::timeout(duration, oauth_persistor.refresh_if_needed())
-                        .await
-                        .map_err(|_| {
-                            anyhow!("timed out handshaking with MCP server after {duration:?}")
-                        })??,
+                    Some(duration) => {
+                        let refresh_persistor = oauth_persistor.clone();
+                        let mut refresh =
+                            tokio::spawn(
+                                async move { refresh_persistor.refresh_if_needed().await },
+                            );
+                        time::timeout(duration, &mut refresh)
+                            .await
+                            .map_err(|_| {
+                                anyhow!("timed out handshaking with MCP server after {duration:?}")
+                            })?
+                            .map_err(|error| {
+                                anyhow!("OAuth refresh task failed for MCP server: {error}")
+                            })??;
+                    }
                     None => oauth_persistor.refresh_if_needed().await?,
                 }
                 remaining_timeout =

codex-rs/rmcp-client/tests/streamable_http_recovery.rs

@@ -2,6 +2,7 @@ mod streamable_http_test_support;
 
 use std::ffi::OsString;
 use std::time::Duration;
+use std::time::Instant;
 
 use codex_config::types::OAuthCredentialsStoreMode;
 use codex_exec_server::Environment;
@@ -17,6 +18,7 @@ use rmcp::transport::auth::OAuthTokenResponse;
 use rmcp::transport::auth::VendorExtraTokenFields;
 use serial_test::serial;
 use tempfile::TempDir;
+use tokio::time::sleep;
 
 use streamable_http_test_support::arm_session_post_failure;
 use streamable_http_test_support::call_echo_tool;
@@ -213,7 +215,97 @@ async fn streamable_http_oauth_refreshes_expired_token_before_initialize() -> an
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
 #[serial(oauth_credentials_env)]
-async fn streamable_http_oauth_refresh_respects_initialize_timeout() -> anyhow::Result<()> {
+async fn streamable_http_oauth_preserves_refresh_token_when_refresh_response_omits_rotation()
+-> anyhow::Result<()> {
+    let (_server, base_url) = spawn_streamable_http_server_with_env(&[
+        ("MCP_EXPECT_BEARER", REFRESHED_ACCESS_TOKEN),
+        ("MCP_EXPECT_REFRESH_TOKEN", VALID_REFRESH_TOKEN),
+        ("MCP_REFRESH_ACCESS_TOKEN", REFRESHED_ACCESS_TOKEN),
+        ("MCP_OMIT_ROTATED_REFRESH_TOKEN", "1"),
+    ])
+    .await?;
+    let codex_home = TempCodexHome::new()?;
+    let server_url = format!("{base_url}/mcp");
+    save_expired_oauth_tokens(&server_url)?;
+
+    let client = RmcpClient::new_streamable_http_client(
+        OAUTH_TEST_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?;
+
+    let credentials = std::fs::read_to_string(codex_home.dir.path().join(".credentials.json"))?;
+    assert!(credentials.contains(REFRESHED_ACCESS_TOKEN));
+    assert!(credentials.contains(VALID_REFRESH_TOKEN));
+    assert!(!credentials.contains(EXPIRED_ACCESS_TOKEN));
+    assert!(!credentials.contains(ROTATED_REFRESH_TOKEN));
+
+    Ok(())
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+#[serial(oauth_credentials_env)]
+async fn streamable_http_oauth_concurrent_initializes_share_refreshed_credentials()
+-> anyhow::Result<()> {
+    let (_server, base_url) = spawn_streamable_http_server_with_env(&[
+        ("MCP_EXPECT_BEARER", REFRESHED_ACCESS_TOKEN),
+        ("MCP_EXPECT_REFRESH_TOKEN", VALID_REFRESH_TOKEN),
+        ("MCP_REFRESH_ACCESS_TOKEN", REFRESHED_ACCESS_TOKEN),
+        ("MCP_ROTATED_REFRESH_TOKEN", ROTATED_REFRESH_TOKEN),
+        ("MCP_REFRESH_TOKEN_MAX_USES", "1"),
+        ("MCP_REFRESH_TOKEN_DELAY_MS", "50"),
+    ])
+    .await?;
+    let _codex_home = TempCodexHome::new()?;
+    let server_url = format!("{base_url}/mcp");
+    save_expired_oauth_tokens(&server_url)?;
+
+    let client_a = RmcpClient::new_streamable_http_client(
+        OAUTH_TEST_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?;
+    let client_b = RmcpClient::new_streamable_http_client(
+        OAUTH_TEST_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?;
+
+    let (initialized_a, initialized_b) =
+        tokio::join!(initialize_client(&client_a), initialize_client(&client_b));
+    initialized_a?;
+    initialized_b?;
+
+    let result_a = call_echo_tool(&client_a, "after-shared-refresh-a").await?;
+    let result_b = call_echo_tool(&client_b, "after-shared-refresh-b").await?;
+    assert_eq!(result_a, expected_echo_result("after-shared-refresh-a"));
+    assert_eq!(result_b, expected_echo_result("after-shared-refresh-b"));
+
+    Ok(())
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+#[serial(oauth_credentials_env)]
+async fn streamable_http_oauth_refresh_timeout_keeps_refresh_running() -> anyhow::Result<()> {
     let (_server, base_url) = spawn_streamable_http_server_with_env(&[
         ("MCP_EXPECT_BEARER", REFRESHED_ACCESS_TOKEN),
         ("MCP_EXPECT_REFRESH_TOKEN", VALID_REFRESH_TOKEN),
@@ -222,7 +314,7 @@ async fn streamable_http_oauth_refresh_respects_initialize_timeout() -> anyhow::
         ("MCP_REFRESH_TOKEN_DELAY_MS", "200"),
     ])
     .await?;
-    let _codex_home = TempCodexHome::new()?;
+    let codex_home = TempCodexHome::new()?;
     let server_url = format!("{base_url}/mcp");
     save_expired_oauth_tokens(&server_url)?;
 
@@ -248,6 +340,99 @@ async fn streamable_http_oauth_refresh_respects_initialize_timeout() -> anyhow::
         "expected initialize timeout, got: {error:#}"
     );
 
+    let credentials_path = codex_home.dir.path().join(".credentials.json");
+    let deadline = Instant::now() + Duration::from_secs(2);
+    loop {
+        let credentials = std::fs::read_to_string(&credentials_path)?;
+        if credentials.contains(REFRESHED_ACCESS_TOKEN)
+            && credentials.contains(ROTATED_REFRESH_TOKEN)
+            && !credentials.contains(EXPIRED_ACCESS_TOKEN)
+        {
+            break;
+        }
+
+        if Instant::now() >= deadline {
+            anyhow::bail!("timed out waiting for background refresh to persist credentials");
+        }
+        sleep(Duration::from_millis(25)).await;
+    }
+
+    Ok(())
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+#[serial(oauth_credentials_env)]
+async fn streamable_http_oauth_refresh_and_initialize_share_timeout_budget() -> anyhow::Result<()> {
+    let (_server, base_url) = spawn_streamable_http_server_with_env(&[
+        ("MCP_EXPECT_BEARER", REFRESHED_ACCESS_TOKEN),
+        ("MCP_EXPECT_REFRESH_TOKEN", VALID_REFRESH_TOKEN),
+        ("MCP_REFRESH_ACCESS_TOKEN", REFRESHED_ACCESS_TOKEN),
+        ("MCP_ROTATED_REFRESH_TOKEN", ROTATED_REFRESH_TOKEN),
+        ("MCP_REFRESH_TOKEN_DELAY_MS", "100"),
+        ("MCP_INITIALIZE_DELAY_MS", "200"),
+    ])
+    .await?;
+    let _codex_home = TempCodexHome::new()?;
+    let server_url = format!("{base_url}/mcp");
+    save_expired_oauth_tokens(&server_url)?;
+
+    let client = RmcpClient::new_streamable_http_client(
+        OAUTH_TEST_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?;
+
+    let error = initialize_client_with_timeout(&client, Some(Duration::from_millis(250)))
+        .await
+        .unwrap_err();
+    assert!(
+        error
+            .to_string()
+            .contains("timed out handshaking with MCP server after 250ms"),
+        "expected initialize timeout, got: {error:#}"
+    );
+
+    Ok(())
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+#[serial(oauth_credentials_env)]
+async fn streamable_http_oauth_refresh_failure_reports_auth_required() -> anyhow::Result<()> {
+    let (_server, base_url) = spawn_streamable_http_server_with_env(&[
+        ("MCP_EXPECT_BEARER", REFRESHED_ACCESS_TOKEN),
+        ("MCP_EXPECT_REFRESH_TOKEN", "different-refresh-token"),
+        ("MCP_REFRESH_ACCESS_TOKEN", REFRESHED_ACCESS_TOKEN),
+        ("MCP_ROTATED_REFRESH_TOKEN", ROTATED_REFRESH_TOKEN),
+    ])
+    .await?;
+    let _codex_home = TempCodexHome::new()?;
+    let server_url = format!("{base_url}/mcp");
+    save_expired_oauth_tokens(&server_url)?;
+
+    let client = RmcpClient::new_streamable_http_client(
+        OAUTH_TEST_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?;
+
+    let error = initialize_client(&client).await.unwrap_err();
+    assert!(
+        error.to_string().contains("Auth required"),
+        "expected auth-required error, got: {error:#}"
+    );
+
     Ok(())
 }