Refine enterprise MCP auth errors and request authority checks (#55691)

## What changed

- Classify token-source failures by credential kind and exchange stage, with sign-in or administrator guidance that omits provider error text. Treat resource authorization failures separately from enterprise identity failures when deciding whether authentication is required.
- Add policy-bound request authority checks that return `RequestNotSent` when policy is updating or unavailable, while allowing admitted responses to finish. Support child authorities that isolate connection revocation and share response draining.
- Recheck account validity after acquiring the credential invalidation lock and check cancellation before deleting a rejected credential.

## Testing

Extend regression coverage for policy suspension during token renewal, completion of admitted requests, recovery with fresh authority, and transport retry classification. Expand provider-error tests to cover `unauthorized_client` and `invalid_client` while checking that credential text is not exposed.

GitOrigin-RevId: c0097adc35cb5fdffde1466e4bc2b5ec78114a4b

codex-rs/codex-mcp/src/ema_auth.rs

@@ -131,7 +131,10 @@ impl EmaAuth {
         self.authority.check()?;
         if !self.manager.account_login_is_current(&self.account).await? {
             self.authority.revoke();
-            bail!(EmaAuthFailure::ReauthenticationRequired);
+            bail!(codex_rmcp_client::EmaTokenSourceError::new(
+                codex_rmcp_client::EmaCredentialKind::Account,
+                EmaAuthFailure::ReauthenticationRequired,
+            ));
         }
         self.authority.check()?;
         Ok(())
@@ -350,16 +353,24 @@ impl EmaTokenSource for EnterpriseTokenSource {
                     })
                 ) && self
                     .credentials
-                    .invalidate_ema_credentials_if_current()
+                    .invalidate_ema_credentials_if_current(async {
+                        self.auth.check_account().await.is_ok()
+                    })
                     .await
                     .is_err()
                 {
                     tracing::warn!("failed to clear rejected enterprise credentials");
                 }
-                Err(match error.downcast::<EmaAuthFailure>() {
-                    Ok(failure) => failure.into(),
-                    Err(_) => anyhow!("enterprise MCP token exchange failed"),
-                })
+                Err(match error.downcast_ref::<EmaAuthFailure>() {
+                    Some(failure) => codex_rmcp_client::EmaTokenSourceError::new(
+                        codex_rmcp_client::EmaCredentialKind::Keyring,
+                        *failure,
+                    ),
+                    None => codex_rmcp_client::EmaTokenSourceError::unexpected(
+                        codex_rmcp_client::EmaCredentialKind::Keyring,
+                    ),
+                }
+                .into())
             }
         }
     }

codex-rs/exec-server/src/client.rs

@@ -701,6 +701,8 @@ pub enum ExecServerError {
     Json(#[from] serde_json::Error),
     #[error("HTTP request failed: {0}")]
     HttpRequest(String),
+    #[error("request not sent: {0}")]
+    RequestNotSent(String),
     #[error("authentication required: {0}")]
     AuthenticationRequired(String),
     #[error("exec-server protocol error: {0}")]

codex-rs/exec-server/src/client/connection_failure.rs

@@ -43,6 +43,7 @@ impl ExecServerError {
             | Self::ProvisioningFailed(_)
             | Self::Json(_)
             | Self::HttpRequest(_)
+            | Self::RequestNotSent(_)
             | Self::Protocol(_)
             | Self::AuthenticationRequired(_)
             | Self::ProvisioningModeConflict { .. }

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

@@ -9,12 +9,14 @@ use url::Host;
 use url::Url;
 
 /// A sanitized enterprise-auth failure that callers may handle without parsing text.
-#[derive(Debug, PartialEq, Eq, thiserror::Error)]
+#[derive(Clone, Copy, Debug, PartialEq, Eq, thiserror::Error)]
 pub enum EmaAuthFailure {
     #[error("invalid_grant")]
     InvalidGrant { grant_source: EmaInvalidGrantSource },
     #[error("insufficient_user_authentication")]
-    InsufficientUserAuthentication,
+    InsufficientUserAuthentication { grant_source: EmaInvalidGrantSource },
+    #[error("enterprise identity rejected the OAuth client")]
+    IdentityClientRejected,
     #[error("enterprise identity requires authentication")]
     ReauthenticationRequired,
 }
@@ -25,6 +27,138 @@ pub enum EmaInvalidGrantSource {
     ResourceAuthorization,
 }
 
+/// The credential source selected for enterprise token acquisition or validation.
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub enum EmaCredentialKind {
+    Account,
+    Keyring,
+}
+
+/// Carries only a credential kind and an already sanitized auth failure to transport.
+/// Unexpected errors deliberately retain neither provider text nor their source chain.
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub enum EmaTokenSourceError {
+    AuthFailure {
+        credential: EmaCredentialKind,
+        failure: EmaAuthFailure,
+    },
+    Unexpected {
+        credential: EmaCredentialKind,
+    },
+}
+
+impl EmaTokenSourceError {
+    pub fn new(credential: EmaCredentialKind, failure: EmaAuthFailure) -> Self {
+        Self::AuthFailure {
+            credential,
+            failure,
+        }
+    }
+
+    pub fn unexpected(credential: EmaCredentialKind) -> Self {
+        Self::Unexpected { credential }
+    }
+
+    fn credential(self) -> EmaCredentialKind {
+        match self {
+            Self::AuthFailure { credential, .. } | Self::Unexpected { credential } => credential,
+        }
+    }
+
+    fn failure(self) -> Option<EmaAuthFailure> {
+        match self {
+            Self::AuthFailure { failure, .. } => Some(failure),
+            Self::Unexpected { .. } => None,
+        }
+    }
+
+    pub fn is_authentication_required(&self) -> bool {
+        match (self.credential(), self.failure()) {
+            (
+                _,
+                Some(
+                    EmaAuthFailure::InvalidGrant {
+                        grant_source: EmaInvalidGrantSource::EnterpriseIdentity,
+                    }
+                    | EmaAuthFailure::InsufficientUserAuthentication {
+                        grant_source: EmaInvalidGrantSource::EnterpriseIdentity,
+                    }
+                    | EmaAuthFailure::ReauthenticationRequired,
+                ),
+            )
+            | (EmaCredentialKind::Account, Some(EmaAuthFailure::IdentityClientRejected)) => true,
+            (
+                _,
+                Some(
+                    EmaAuthFailure::InvalidGrant {
+                        grant_source: EmaInvalidGrantSource::ResourceAuthorization,
+                    }
+                    | EmaAuthFailure::InsufficientUserAuthentication {
+                        grant_source: EmaInvalidGrantSource::ResourceAuthorization,
+                    },
+                )
+                | None,
+            )
+            | (EmaCredentialKind::Keyring, Some(EmaAuthFailure::IdentityClientRejected)) => false,
+        }
+    }
+}
+
+impl std::fmt::Display for EmaTokenSourceError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        let message = match (self.credential(), self.failure()) {
+            (
+                _,
+                Some(EmaAuthFailure::InsufficientUserAuthentication {
+                    grant_source: EmaInvalidGrantSource::ResourceAuthorization,
+                }),
+            ) => {
+                "The MCP authorization server requires additional user authentication; contact your administrator"
+            }
+            (EmaCredentialKind::Keyring, Some(EmaAuthFailure::IdentityClientRejected)) => {
+                "The enterprise IdP rejected the configured MCP OAuth client; contact your administrator"
+            }
+            (
+                EmaCredentialKind::Account,
+                Some(
+                    EmaAuthFailure::InvalidGrant {
+                        grant_source: EmaInvalidGrantSource::EnterpriseIdentity,
+                    }
+                    | EmaAuthFailure::InsufficientUserAuthentication {
+                        grant_source: EmaInvalidGrantSource::EnterpriseIdentity,
+                    }
+                    | EmaAuthFailure::IdentityClientRejected,
+                ),
+            ) => "Complete enterprise sign-in for this MCP server",
+            (EmaCredentialKind::Account, Some(EmaAuthFailure::ReauthenticationRequired)) => {
+                "Sign in to your Codex account to restore enterprise MCP access"
+            }
+            (
+                EmaCredentialKind::Keyring,
+                Some(
+                    EmaAuthFailure::InvalidGrant {
+                        grant_source: EmaInvalidGrantSource::EnterpriseIdentity,
+                    }
+                    | EmaAuthFailure::InsufficientUserAuthentication {
+                        grant_source: EmaInvalidGrantSource::EnterpriseIdentity,
+                    }
+                    | EmaAuthFailure::ReauthenticationRequired,
+                ),
+            ) => "Complete enterprise sign-in to restore MCP access",
+            (
+                _,
+                Some(EmaAuthFailure::InvalidGrant {
+                    grant_source: EmaInvalidGrantSource::ResourceAuthorization,
+                }),
+            ) => "The MCP authorization server rejected the enterprise grant: invalid_grant",
+            (_, None) => "enterprise MCP token exchange failed",
+        };
+        f.write_str(message)
+    }
+}
+
+impl std::error::Error for EmaTokenSourceError {}
+
 pub(crate) fn ema_reauthentication_required(message: &'static str) -> anyhow::Error {
     anyhow::Error::new(EmaAuthFailure::ReauthenticationRequired).context(message)
 }

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

@@ -127,9 +127,22 @@ fn exchange_error(error: EmaError) -> anyhow::Error {
                 grant_source: EmaInvalidGrantSource::ResourceAuthorization,
             })
         }
-        EmaError::InsufficientUserAuthentication(_) => {
-            Some(EmaAuthFailure::InsufficientUserAuthentication)
+        EmaError::InsufficientUserAuthentication(EmaExchangeStage::IdentityProvider) => {
+            Some(EmaAuthFailure::InsufficientUserAuthentication {
+                grant_source: EmaInvalidGrantSource::EnterpriseIdentity,
+            })
         }
+        EmaError::InsufficientUserAuthentication(EmaExchangeStage::ResourceAuthorizationServer) => {
+            Some(EmaAuthFailure::InsufficientUserAuthentication {
+                grant_source: EmaInvalidGrantSource::ResourceAuthorization,
+            })
+        }
+        EmaError::OAuthRejected {
+            stage: EmaExchangeStage::IdentityProvider,
+            code: "unauthorized_client" | "invalid_client",
+            ..
+        } => Some(EmaAuthFailure::IdentityClientRejected),
+        // Future SDK stages must not invalidate the shared enterprise credential.
         _ => None,
     };
     match failure {

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

@@ -332,6 +332,8 @@ async fn provider_errors_preserve_credential_policy_without_reflecting_credentia
             SENTINEL,
             "invalid_grant",
             "insufficient_user_authentication",
+            "unauthorized_client",
+            "invalid_client",
             "malformed",
         ] {
             let server = MockServer::builder().start().await;
@@ -367,7 +369,12 @@ async fn provider_errors_preserve_credential_policy_without_reflecting_credentia
             let expected = match code {
                 "invalid_grant" => Some(EmaAuthFailure::InvalidGrant { grant_source }),
                 "insufficient_user_authentication" => {
-                    Some(EmaAuthFailure::InsufficientUserAuthentication)
+                    Some(EmaAuthFailure::InsufficientUserAuthentication { grant_source })
+                }
+                "unauthorized_client" | "invalid_client"
+                    if grant_source == EmaInvalidGrantSource::EnterpriseIdentity =>
+                {
+                    Some(EmaAuthFailure::IdentityClientRejected)
                 }
                 _ => None,
             };

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

@@ -42,6 +42,7 @@ pub struct EmaRequestAuthority {
     revoked: CancellationToken,
     is_current: Arc<dyn Fn() -> bool + Send + Sync>,
     responses: Arc<RwLock<()>>,
+    policy: Option<codex_config::CloudConfigBundleBinding>,
 }
 
 impl Default for EmaRequestAuthority {
@@ -61,6 +62,24 @@ impl EmaRequestAuthority {
             revoked,
             is_current: Arc::new(is_current),
             responses: Arc::default(),
+            policy: None,
+        }
+    }
+
+    /// Policy changes stop new dispatches without cancelling already admitted responses.
+    pub fn with_policy(mut self, policy: Option<codex_config::CloudConfigBundleBinding>) -> Self {
+        self.policy = policy;
+        self
+    }
+
+    /// Isolate one connection's credential failure while sharing the owner's response drain.
+    pub fn child(&self) -> Self {
+        let owner = self.clone();
+        Self {
+            revoked: self.revoked.child_token(),
+            is_current: Arc::new(move || !owner.is_revoked()),
+            responses: Arc::clone(&self.responses),
+            policy: self.policy.clone(),
         }
     }
 
@@ -82,6 +101,19 @@ impl EmaRequestAuthority {
         Ok(())
     }
 
+    pub fn check_new_request(&self) -> Result<(), ExecServerError> {
+        self.check()?;
+        if self.policy.as_ref().is_some_and(|binding| {
+            binding.read().status != codex_config::CloudConfigBundleBindingStatus::Current
+        }) {
+            return Err(ExecServerError::RequestNotSent(
+                "enterprise MCP policy is updating or unavailable; retry after it is applied"
+                    .into(),
+            ));
+        }
+        Ok(())
+    }
+
     /// Wait until response futures admitted before revocation complete or are cancelled.
     pub async fn quiesce(&self) {
         let _responses = self.responses.clone().write_owned().await;
@@ -89,7 +121,7 @@ impl EmaRequestAuthority {
 
     async fn acquire(&self) -> Result<OwnedRwLockReadGuard<()>, ExecServerError> {
         let response = self.responses.clone().read_owned().await;
-        self.check()?;
+        self.check_new_request()?;
         Ok(response)
     }
 
@@ -180,7 +212,12 @@ impl<S: EmaTokenSource> EmaAuthenticatedHttpClient<S> {
             if let Some(value) = current_token() {
                 value
             } else {
-                let renewed = self.source.access_token().await.map_err(auth_error)?;
+                let renewed = self.source.access_token().await.map_err(|error| {
+                    if let Err(refusal) = self.authority.check_new_request() {
+                        return refusal;
+                    }
+                    auth_error(error)
+                })?;
                 let token = CachedToken::new(renewed);
                 if Instant::now() >= token.refresh_at {
                     return Err(ExecServerError::HttpRequest(
@@ -234,6 +271,13 @@ impl<S: EmaTokenSource> EmaAuthenticatedHttpClient<S> {
 }
 
 fn auth_error(error: anyhow::Error) -> ExecServerError {
+    if let Some(error) = error.downcast_ref::<crate::EmaTokenSourceError>() {
+        return if error.is_authentication_required() {
+            ExecServerError::AuthenticationRequired(error.to_string())
+        } else {
+            ExecServerError::HttpRequest(error.to_string())
+        };
+    }
     if is_authentication_required_error(&error) {
         ExecServerError::AuthenticationRequired(
             "enterprise MCP authentication is required".to_string(),
@@ -257,7 +301,7 @@ impl<S: EmaTokenSource> HttpClient for EmaAuthenticatedHttpClient<S> {
             let request = async {
                 let _authority = self.authority.acquire().await?;
                 let token = self.authenticate(&mut params).await?;
-                self.authority.check()?;
+                self.authority.check_new_request()?;
                 if let Some(deadline) = deadline {
                     params.timeout_ms = Some(remaining_timeout(deadline)?);
                 }
@@ -286,7 +330,7 @@ impl<S: EmaTokenSource> HttpClient for EmaAuthenticatedHttpClient<S> {
             let request = async {
                 let _authority = self.authority.acquire().await?;
                 let token = self.authenticate(&mut params).await?;
-                self.authority.check()?;
+                self.authority.check_new_request()?;
                 if let Some(deadline) = deadline {
                     params.timeout_ms = Some(remaining_timeout(deadline)?);
                 }

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

@@ -1,5 +1,8 @@
 //! Authentication, deadlines, and concurrent responses share the same request invariants.
 
+use codex_config::CloudConfigBundlePolicy;
+use codex_config::CloudConfigBundleSnapshot;
+use codex_config::test_support::CloudConfigBundleFixture;
 use pretty_assertions::assert_eq;
 use tokio::sync::mpsc;
 use tokio::sync::oneshot;
@@ -206,7 +209,9 @@ async fn credential_failures_are_classified_before_dispatch() -> anyhow::Result<
             EmaAuthFailure::InvalidGrant {
                 grant_source: EmaInvalidGrantSource::EnterpriseIdentity,
             },
-            EmaAuthFailure::InsufficientUserAuthentication,
+            EmaAuthFailure::InsufficientUserAuthentication {
+                grant_source: crate::EmaInvalidGrantSource::EnterpriseIdentity,
+            },
             EmaAuthFailure::ReauthenticationRequired,
         ] {
             let (client, source, params, mut requests) = controlled_fixture();
@@ -385,3 +390,69 @@ async fn authentication_time_is_removed_from_the_inner_timeout() -> anyhow::Resu
     }
     Ok(())
 }
+
+#[tokio::test]
+async fn policy_suspension_requires_fresh_authority_without_cancelling_admitted_request()
+-> anyhow::Result<()> {
+    let (mut client, source, params, mut requests) = controlled_fixture();
+    let policy = CloudConfigBundlePolicy::default();
+    let mut snapshot = CloudConfigBundleSnapshot {
+        bundle: Ok(None),
+        binding: None,
+    };
+    policy.publish_snapshot(&mut snapshot);
+    Arc::get_mut(&mut client)
+        .expect("single client owner")
+        .authority = EmaRequestAuthority::default().with_policy(snapshot.binding.take());
+
+    let pending = RequestMode::Buffered.start(Arc::clone(&client), params.clone());
+    let dispatched = requests.recv().await.expect("admitted request");
+    *client.cached_token() = None;
+    let (gate, waiting, release) = CredentialGate::new();
+    *source.credential_gate.lock().unwrap() = Some(gate);
+    let renewal = RequestMode::Streaming.start(Arc::clone(&client), params.clone());
+    waiting.await?;
+    let next = CloudConfigBundleFixture::enterprise_config("model = 'managed'").into_bundle();
+    policy.observe_remote_bundle(&next);
+    *source.renewal_error.lock().unwrap() =
+        Some(anyhow::anyhow!("exchange refused after policy changed"));
+    release.send(()).expect("release renewal");
+    assert!(matches!(
+        renewal.await?,
+        Err(ExecServerError::RequestNotSent(_))
+    ));
+    assert!(matches!(
+        client.http_request(params.clone()).await,
+        Err(ExecServerError::RequestNotSent(message)) if message.contains("retry")
+    ));
+    assert!(matches!(
+        requests.try_recv(),
+        Err(mpsc::error::TryRecvError::Empty)
+    ));
+    dispatched
+        .response
+        .send(200)
+        .expect("complete admitted request");
+    assert_eq!(pending.await??.status, 200);
+
+    snapshot.bundle = Ok(Some(next));
+    policy.publish_snapshot(&mut snapshot);
+    assert!(matches!(
+        client.http_request(params).await,
+        Err(ExecServerError::RequestNotSent(_))
+    ));
+    let (mut fresh_client, _source, fresh_params, mut fresh_requests) = controlled_fixture();
+    Arc::get_mut(&mut fresh_client)
+        .expect("single client owner")
+        .authority = EmaRequestAuthority::default().with_policy(snapshot.binding);
+    let pending = RequestMode::Buffered.start(fresh_client, fresh_params);
+    fresh_requests
+        .recv()
+        .await
+        .expect("request after recovery")
+        .response
+        .send(200)
+        .expect("complete recovered request");
+    assert_eq!(pending.await??.status, 200);
+    Ok(())
+}

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

@@ -48,7 +48,9 @@ pub use codex_protocol::protocol::McpAuthStatus;
 pub use ema_auth::EmaAuthTokenExchangeRequest;
 pub use ema_auth::exchange_ema_auth_token;
 pub use ema_auth_policy::EmaAuthFailure;
+pub use ema_auth_policy::EmaCredentialKind;
 pub use ema_auth_policy::EmaInvalidGrantSource;
+pub use ema_auth_policy::EmaTokenSourceError;
 pub use ema_auth_policy::validate_ema_auth_resource;
 pub use ema_exchange::EmaAccessToken;
 pub use ema_http_client::EmaAuthenticatedHttpClient;

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

@@ -1,6 +1,8 @@
 //! Pin enterprise credentials to a non-secret version advanced by cooperating writers.
 //! Exchanges reread the pinned keyring record and reject changes made outside this protocol.
 
+use std::future::Future;
+
 use anyhow::Result;
 use anyhow::anyhow;
 use anyhow::bail;
@@ -156,14 +158,18 @@ impl EmaCredentialSnapshot {
     }
 
     /// Exclusively lock and delete a rejected grant only if its pinned record is still current.
-    pub async fn invalidate_ema_credentials_if_current(&self) -> Result<()> {
-        self.invalidate_ema_credentials_if_current_in(&DefaultKeyringStore)
+    pub async fn invalidate_ema_credentials_if_current(
+        &self,
+        is_current: impl Future<Output = bool> + Send,
+    ) -> Result<()> {
+        self.invalidate_ema_credentials_if_current_in(&DefaultKeyringStore, is_current)
             .await
     }
 
     async fn invalidate_ema_credentials_if_current_in<K: KeyringStore + Clone + 'static>(
         &self,
         keyring_store: &K,
+        is_current: impl Future<Output = bool> + Send,
     ) -> Result<()> {
         let credential_lock = RefreshCredentialLock::acquire_for_server(
             &self.snapshot.credentials.server_name,
@@ -172,6 +178,11 @@ impl EmaCredentialSnapshot {
         .with_subscriber(tracing::subscriber::NoSubscriber::default())
         .await
         .map_err(|_| anyhow!("failed to lock enterprise credentials"))?;
+        if !is_current.await {
+            return Ok(());
+        }
+        let cancelled = tokio_util::sync::CancellationToken::new();
+        let _cancel_on_drop = cancelled.clone().drop_guard();
         let pinned = self.clone();
         let keyring_store = keyring_store.clone();
         tokio::task::spawn_blocking(move || {
@@ -193,7 +204,7 @@ impl EmaCredentialSnapshot {
                         .load(&keyring_store, &previous.server_name, &previous.url)?
                 {
                     latest.token_response.0.set_expires_in(None);
-                    if latest == *previous {
+                    if latest == *previous && !cancelled.is_cancelled() {
                         invalidate_enterprise_credential_version(
                             &previous.server_name,
                             &previous.url,

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

@@ -78,8 +78,9 @@ async fn invalidation_rereads_under_exclusive_lock_and_preserves_newer_grants()
             version: versions.replace()?,
         };
         store.save(&keyring, &tokens.server_name, &tokens)?;
-        let mut invalidation =
-            Box::pin(snapshot.invalidate_ema_credentials_if_current_in(&keyring));
+        let mut invalidation = Box::pin(
+            snapshot.invalidate_ema_credentials_if_current_in(&keyring, std::future::ready(true)),
+        );
         assert!(invalidation.as_mut().now_or_never().is_none());
         assert_eq!(
             store.load(&keyring, &tokens.server_name, &tokens.url)?,
@@ -331,7 +332,7 @@ async fn enterprise_credential_errors_and_keyring_traces_hide_account_key() -> R
         .0
         .set_error(&key, keyring::Error::Invalid("account".into(), key.clone()));
     let error = pinned
-        .invalidate_ema_credentials_if_current_in(&keyring)
+        .invalidate_ema_credentials_if_current_in(&keyring, std::future::ready(true))
         .await
         .unwrap_err();
     assert!(!format!("{error} {error:?} {error:#}").contains(sentinel));
@@ -342,7 +343,7 @@ async fn enterprise_credential_errors_and_keyring_traces_hide_account_key() -> R
         .clone()
         .pin_ema_credentials_in(&successful_keyring)
         .await?
-        .invalidate_ema_credentials_if_current_in(&successful_keyring)
+        .invalidate_ema_credentials_if_current_in(&successful_keyring, std::future::ready(true))
         .await?;
     assert_eq!(successful_keyring.0.saved_value(&key), None);
 
@@ -357,7 +358,7 @@ async fn enterprise_credential_errors_and_keyring_traces_hide_account_key() -> R
             .expect("lock must fail"),
         pinned.acquire_current_ema_credentials().await.unwrap_err(),
         pinned
-            .invalidate_ema_credentials_if_current_in(&keyring)
+            .invalidate_ema_credentials_if_current_in(&keyring, std::future::ready(true))
             .await
             .unwrap_err(),
     ] {

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

@@ -18,6 +18,9 @@ use crate::rmcp_client::ClientOperationError;
 /// transport error, which is not part of the standard error source chain.
 pub fn is_authentication_required_error(error: &Error) -> bool {
     error.chain().any(|source| {
+        if let Some(error) = source.downcast_ref::<crate::EmaTokenSourceError>() {
+            return error.is_authentication_required();
+        }
         source
             .downcast_ref::<AuthError>()
             .is_some_and(auth_error_requires_authentication)
@@ -27,11 +30,17 @@ pub fn is_authentication_required_error(error: &Error) -> bool {
                     EmaAuthFailure::InvalidGrant {
                         grant_source: EmaInvalidGrantSource::EnterpriseIdentity,
                     }
-                    | EmaAuthFailure::InsufficientUserAuthentication
+                    | EmaAuthFailure::InsufficientUserAuthentication {
+                        grant_source: EmaInvalidGrantSource::EnterpriseIdentity,
+                    }
+                    | EmaAuthFailure::IdentityClientRejected
                     | EmaAuthFailure::ReauthenticationRequired => true,
                     // Rejecting an ID-JAG does not invalidate the enterprise login itself.
                     EmaAuthFailure::InvalidGrant {
                         grant_source: EmaInvalidGrantSource::ResourceAuthorization,
+                    }
+                    | EmaAuthFailure::InsufficientUserAuthentication {
+                        grant_source: EmaInvalidGrantSource::ResourceAuthorization,
                     } => false,
                 })
             || source

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

@@ -31,6 +31,9 @@ fn retryable_initialize_error_includes_discovery_and_initialized_notification_co
 #[test]
 fn streamable_http_retry_classifies_transport_and_auth_errors() {
     let errors = [
+        StreamableHttpError::Client(StreamableHttpClientAdapterError::HttpRequest(
+            ExecServerError::RequestNotSent("enterprise policy is updating".to_string()),
+        )),
         StreamableHttpError::Client(StreamableHttpClientAdapterError::HttpRequest(
             ExecServerError::HttpRequest("error sending request for url".to_string()),
         )),
@@ -66,7 +69,9 @@ fn streamable_http_retry_classifies_transport_and_auth_errors() {
 
     assert_eq!(
         errors.map(|error| RmcpClient::is_retryable_streamable_http_error(&error)),
-        [true, true, true, false, true, false, true, false, false],
+        [
+            false, true, true, true, false, true, false, true, false, false
+        ],
     );
 }