Reload remote executor credentials safely across token rotation (#58251)

## Why

ChatGPT credentials can rotate while a remote executor is idle. Registry requests need to follow stored rotations without continuing after logout or switching the executor's identity. Concurrent refreshes must also avoid overwriting newer credentials.

## What changed

- Reload and validate ChatGPT credentials before registry requests, pinning the credential mode, account, user, and workspace policy. Reuse cached metadata for unchanged personal access tokens.
- Retry registration when temporary credential resolution failures prevent a request from being sent. Bound credential resolution and block reuse of a token generation whose refresh outcome is unknown.
- Restrict supported authentication modes before executor startup loads credentials or cloud configuration.
- Serialize persistent credential writes across cooperating processes and compare stored credentials before committing refreshed tokens.
- Replace `auth.json` atomically so overlapping readers see a complete document, preserving symlink targets and protecting file permissions, including Windows access controls.

## Testing

Add coverage for credential rotation, logout, identity changes, cancelled refresh callers, temporary metadata failures, registration retries, concurrent credential writes, and atomic file replacement.

GitOrigin-RevId: 6b210944bf8eba2f45bc27113f901d2c3cb06b48

codex-rs/exec-server-cli/Cargo.toml

@@ -29,6 +29,7 @@ codex-exec-server = { workspace = true }
 codex-http-client = { workspace = true }
 codex-login = { workspace = true }
 codex-model-provider = { workspace = true }
+codex-protocol = { workspace = true }
 codex-sandboxing = { workspace = true }
 codex-utils-absolute-path = { workspace = true }
 codex-utils-cli = { workspace = true }
@@ -47,10 +48,13 @@ tikv-jemallocator = { workspace = true }
 
 [dev-dependencies]
 assert_matches = { workspace = true }
-codex-protocol = { workspace = true }
+chrono = { workspace = true }
 codex-utils-cargo-bin = { workspace = true }
 codex-utils-process = { workspace = true }
+futures = { workspace = true }
 insta = { workspace = true }
+jsonwebtoken = { workspace = true }
 pretty_assertions = { workspace = true }
 tempfile = { workspace = true }
 tokio = { workspace = true, features = ["io-util", "process"] }
+wiremock = { workspace = true }

codex-rs/exec-server-cli/src/config.rs

@@ -1,7 +1,7 @@
 //! Loads executor startup configuration and remote authentication without changing CLI policy.
 
 use codex_app_server::in_process::EmbeddedNetworkPolicy;
-use codex_cloud_config::chatgpt_config_bundle_loader_for_storage;
+use codex_cloud_config::chatgpt_config_bundle_loader;
 use codex_config::LoaderOverrides;
 use codex_core::config::ConfigBuilder;
 use codex_core::config::ConfigLoadOptions;
@@ -11,10 +11,20 @@ use codex_core::config::load_config_toml_with_layer_stack;
 use codex_login::AuthManager;
 use codex_login::CodexAuth;
 use codex_login::read_codex_access_token_from_env;
+use codex_protocol::auth::AuthMode;
 use codex_utils_absolute_path::AbsolutePathBuf;
 use codex_utils_cli::CliConfigOverrides;
 use std::sync::Arc;
 
+use crate::exec_server_auth;
+
+const REMOTE_AUTH_MODES: &[AuthMode] = &[
+    AuthMode::ApiKey,
+    AuthMode::Chatgpt,
+    AuthMode::ChatgptAuthTokens,
+    AuthMode::PersonalAccessToken,
+];
+
 pub(super) async fn load_exec_server_remote_auth_provider(
     config: &codex_core::config::Config,
     base_url: &str,
@@ -53,11 +63,16 @@ pub(super) async fn load_exec_server_remote_auth_provider(
         validate_api_key_remote_host(base_url)?;
     }
 
-    if auth.is_chatgpt_auth() || auth_manager.is_workload_identity_selected() {
+    if auth_manager.is_workload_identity_selected() {
         Ok(codex_model_provider::auth_provider_from_auth_manager(
             auth_manager,
             &auth,
         ))
+    } else if auth.is_chatgpt_auth() {
+        Ok(exec_server_auth::chatgpt_auth_provider(
+            auth_manager,
+            &auth,
+        )?)
     } else {
         Ok(codex_model_provider::auth_provider_from_auth(&auth))
     }
@@ -129,12 +144,25 @@ pub(super) async fn load_exec_server_config(
             },
         )
         .await?;
-        let bootstrap_auth_config = bootstrap_auth_config(&codex_home, &bootstrap_config)?;
-        let cloud_config_bundle = chatgpt_config_bundle_loader_for_storage(
-            network_policy.bind_bootstrap_auth(bootstrap_auth_config),
+        let bootstrap_auth_config = network_policy
+            .bind_bootstrap_auth(bootstrap_auth_config(&codex_home, &bootstrap_config)?);
+        let auth_manager = AuthManager::shared_from_auth_config_with_auth_modes(
+            bootstrap_auth_config.clone(),
             /*enable_codex_api_key_env*/ true,
+            REMOTE_AUTH_MODES.to_vec(),
         )
         .await?;
+        let cloud_config_bundle = chatgpt_config_bundle_loader(
+            auth_manager,
+            bootstrap_auth_config
+                .chatgpt_base_url
+                .unwrap_or_else(|| "https://chatgpt.com/backend-api/".to_string()),
+            bootstrap_auth_config.codex_home,
+            bootstrap_auth_config
+                .auth_route_config
+                .http_client_factory()
+                .clone(),
+        );
         builder = builder.cloud_config_bundle(cloud_config_bundle);
     }
     let mut config = builder.build().await?;
@@ -146,8 +174,12 @@ async fn load_exec_server_remote_auth(
     config: &codex_core::config::Config,
     missing_auth_error: &'static str,
 ) -> anyhow::Result<(Arc<AuthManager>, codex_login::CodexAuth)> {
-    let auth_manager =
-        AuthManager::shared_from_config(config, /*enable_codex_api_key_env*/ true).await?;
+    let auth_manager = AuthManager::shared_from_config_with_auth_modes(
+        config,
+        /*enable_codex_api_key_env*/ true,
+        REMOTE_AUTH_MODES.to_vec(),
+    )
+    .await?;
 
     let auth = match auth_manager.auth().await {
         Some(auth) => auth,

codex-rs/exec-server-cli/src/exec_server_auth.rs

@@ -1,10 +1,17 @@
-//! CLI-only glue between Codex authentication and generic AWS signing.
+//! Request-time authentication for remote executors, including AWS signing.
 //!
-//! Keeping this adapter here leaves `codex-api` and `codex-aws-auth` independent.
+//! ChatGPT credentials may rotate while an executor is idle. Reload them for
+//! registry requests while retaining the identity that registered the executor.
+//! Failed credential reads block headers until revalidation, without counting as logout.
 
 use std::sync::Arc;
+use std::sync::Mutex;
+use std::sync::atomic::AtomicBool;
+use std::sync::atomic::Ordering;
+use std::time::Duration;
 
 use codex_api::AuthError;
+use codex_api::AuthHeadersFuture;
 use codex_api::AuthProvider;
 use codex_api::SharedAuthProvider;
 use codex_aws_auth::AwsAuthConfig;
@@ -14,7 +21,221 @@ use codex_aws_auth::AwsRequestToSign;
 use codex_http_client::Request;
 use codex_http_client::RequestBody;
 use codex_http_client::RequestCompression;
+use codex_login::AuthManager;
+use codex_login::CodexAuth;
 use http::HeaderMap;
+use tokio::sync::Semaphore;
+
+/// Follows stored credential rotation without allowing an executor's owner to change.
+pub(super) fn chatgpt_auth_provider(
+    auth_manager: Arc<AuthManager>,
+    expected_auth: &CodexAuth,
+) -> Result<SharedAuthProvider, AuthError> {
+    auth_manager
+        .restrict_auth_modes(vec![expected_auth.auth_mode()])
+        .map_err(|_| AuthError::Build("executor credential policy is unavailable".to_string()))?;
+    let account_id = expected_auth.get_account_id();
+    let user_id = expected_auth.get_chatgpt_user_id();
+    let identity_complete = account_id.as_ref().is_some_and(|id| !id.trim().is_empty())
+        && user_id.as_ref().is_some_and(|id| !id.trim().is_empty());
+    let unidentified_headers = if identity_complete {
+        None
+    } else {
+        Some(codex_model_provider::auth_provider_from_auth(expected_auth).to_auth_headers()?)
+    };
+    Ok(Arc::new(ReloadingChatgptAuthProvider {
+        managed: codex_model_provider::auth_provider_from_auth_manager(
+            Arc::clone(&auth_manager),
+            expected_auth,
+        ),
+        auth_manager,
+        account_id,
+        user_id,
+        workspace_account: expected_auth.is_workspace_account(),
+        fedramp_account: expected_auth.is_fedramp_account(),
+        unidentified_headers,
+        resolution_lock: Arc::new(Semaphore::new(/*permits*/ 1)),
+        uncertain_refresh: Arc::new(Mutex::new(None)),
+        headers_valid: Arc::new(AtomicBool::new(true)),
+    }))
+}
+
+#[derive(Clone)]
+struct ReloadingChatgptAuthProvider {
+    managed: SharedAuthProvider,
+    auth_manager: Arc<AuthManager>,
+    account_id: Option<String>,
+    user_id: Option<String>,
+    workspace_account: bool,
+    fedramp_account: bool,
+    unidentified_headers: Option<HeaderMap>,
+    resolution_lock: Arc<Semaphore>,
+    uncertain_refresh: Arc<Mutex<Option<codex_login::TokenData>>>,
+    headers_valid: Arc<AtomicBool>,
+}
+
+impl ReloadingChatgptAuthProvider {
+    fn checked_headers(&self, auth: Option<CodexAuth>) -> Result<HeaderMap, AuthError> {
+        let auth = auth
+            .filter(|auth| {
+                auth.is_chatgpt_auth()
+                    && auth.get_account_id() == self.account_id
+                    && auth.get_chatgpt_user_id() == self.user_id
+                    && auth.is_workspace_account() == self.workspace_account
+                    && auth.is_fedramp_account() == self.fedramp_account
+            })
+            .ok_or_else(|| AuthError::Build("executor authentication changed".to_string()))?;
+        let token = auth.get_token().map_err(|_| {
+            AuthError::Build("executor credential has no usable bearer".to_string())
+        })?;
+        if token.trim().is_empty() {
+            return Err(AuthError::Build(
+                "executor credential has an empty bearer".to_string(),
+            ));
+        }
+        let uncertain = self.uncertain_refresh.lock().map_err(|_| {
+            AuthError::Transient("executor authentication is unavailable".to_string())
+        })?;
+        if uncertain
+            .as_ref()
+            .is_some_and(|previous| auth.get_token_data().as_ref().ok() == Some(previous))
+        {
+            return Err(AuthError::Transient(
+                "executor refresh outcome is unknown; replace or renew stored credentials"
+                    .to_string(),
+            ));
+        }
+        let headers = codex_model_provider::auth_provider_from_auth(&auth).to_auth_headers()?;
+        if self
+            .unidentified_headers
+            .as_ref()
+            .is_some_and(|expected| expected != &headers)
+        {
+            return Err(AuthError::Build(
+                "executor credential rotation requires a complete account and user identity"
+                    .to_string(),
+            ));
+        }
+        Ok(headers)
+    }
+}
+
+impl AuthProvider for ReloadingChatgptAuthProvider {
+    fn add_auth_headers(&self, headers: &mut HeaderMap) -> Result<(), AuthError> {
+        if !self.headers_valid.load(Ordering::Acquire) {
+            return Err(AuthError::Transient(
+                "executor credentials have not been successfully revalidated".to_string(),
+            ));
+        }
+        headers.extend(self.checked_headers(self.auth_manager.auth_cached())?);
+        Ok(())
+    }
+
+    fn apply_auth_headers(&self, mut headers: HeaderMap) -> AuthHeadersFuture<'_> {
+        Box::pin(async move {
+            headers.extend(self.resolve(/*rejected_headers*/ None).await?);
+            Ok(headers)
+        })
+    }
+
+    fn refresh_after_unauthorized(
+        &self,
+        rejected_headers: HeaderMap,
+    ) -> codex_api::AuthRefreshFuture<'_> {
+        Box::pin(async move {
+            self.resolve(Some(rejected_headers.clone()))
+                .await
+                .is_ok_and(|headers| {
+                    headers
+                        .iter()
+                        .any(|(name, value)| rejected_headers.get(name) != Some(value))
+                })
+        })
+    }
+}
+
+impl ReloadingChatgptAuthProvider {
+    async fn resolve(&self, rejected_headers: Option<HeaderMap>) -> Result<HeaderMap, AuthError> {
+        let guard = Arc::clone(&self.resolution_lock)
+            .acquire_owned()
+            .await
+            .map_err(|_| {
+                AuthError::Transient("executor authentication is unavailable".to_string())
+            })?;
+        let provider = self.clone();
+        // A failed read leaves AuthManager's last known owner intact. Keep those
+        // cached credentials unavailable to synchronous callers until revalidated.
+        provider.headers_valid.store(false, Ordering::Release);
+        // Outlive a cancelled handshake, but never hold the permit indefinitely.
+        // A timed-out OAuth exchange may already have rotated the token: quarantine
+        // that generation until another login/refresh supplies different credentials.
+        tokio::spawn(async move {
+            let _guard = guard;
+            let mut attempted_refresh = None;
+            let result = tokio::time::timeout(Duration::from_secs(30), async {
+                provider
+                    .auth_manager
+                    .reload_with_cached_auth()
+                    .await
+                    .map_err(|_| {
+                        AuthError::Transient(
+                            "executor credential storage or hydration is unavailable".to_string(),
+                        )
+                    })?;
+                let auth = provider.auth_manager.auth_cached();
+                let current = provider.checked_headers(auth.clone())?;
+                if provider.unidentified_headers.is_some() {
+                    return Ok(current);
+                }
+                attempted_refresh = auth.as_ref().and_then(|auth| auth.get_token_data().ok());
+                if let Some(rejected) = rejected_headers {
+                    if !current
+                        .iter()
+                        .any(|(name, value)| rejected.get(name) != Some(value))
+                    {
+                        provider.managed.refresh_after_unauthorized(rejected).await;
+                    }
+                } else {
+                    provider.auth_manager.auth().await;
+                }
+                // A later storage timeout is not an interrupted OAuth exchange.
+                attempted_refresh = None;
+                // auth() deliberately falls back to a snapshot on refresh errors.
+                // Re-read the selected source before dispatch: logout, replacement,
+                // or failed hydration must never authorize that fallback snapshot.
+                provider
+                    .auth_manager
+                    .reload_with_cached_auth()
+                    .await
+                    .map_err(|_| {
+                        AuthError::Transient(
+                            "executor credential storage or hydration is unavailable".to_string(),
+                        )
+                    })?;
+                provider.checked_headers(provider.auth_manager.auth_cached())
+            })
+            .await;
+            let result = match result {
+                Ok(result) => result,
+                Err(_) => {
+                    if let Some(attempted) = attempted_refresh
+                        && let Ok(mut uncertain) = provider.uncertain_refresh.lock()
+                    {
+                        *uncertain = Some(attempted);
+                    }
+                    Err(AuthError::Transient(
+                        "executor credential resolution timed out; retry after repairing credentials"
+                            .to_string(),
+                    ))
+                }
+            };
+            provider.headers_valid.store(result.is_ok(), Ordering::Release);
+            result
+        })
+        .await
+        .map_err(|_| AuthError::Transient("executor authentication task failed".to_string()))?
+    }
+}
 
 /// Creates a SigV4 provider, preferring an explicit profile over the default credential chain.
 pub(super) async fn aws_sigv4_auth_provider(

codex-rs/exec-server-cli/src/exec_server_auth_tests.rs

@@ -1,14 +1,564 @@
 #![cfg(test)]
 
+use std::path::Path;
 use std::time::Duration;
 
 use codex_aws_auth::AwsAccessKeys;
+use codex_login::AuthCredentialsStoreMode;
+use codex_login::AuthKeyringBackendKind;
 use http::HeaderValue;
 use http::Method;
 use pretty_assertions::assert_eq;
+use tempfile::TempDir;
+use wiremock::Mock;
+use wiremock::MockServer;
+use wiremock::ResponseTemplate;
+use wiremock::matchers::header;
+use wiremock::matchers::method;
+use wiremock::matchers::path;
 
 use super::*;
 
+fn write_chatgpt_auth(codex_home: &Path, token: &str, account: &str) {
+    write_chatgpt_auth_with_claims(
+        codex_home,
+        token,
+        Some(account),
+        serde_json::json!({"chatgpt_user_id": "user", "chatgpt_plan_type": "pro"}),
+    );
+}
+
+fn write_chatgpt_auth_with_claims(
+    codex_home: &Path,
+    token: &str,
+    account: Option<&str>,
+    claims: serde_json::Value,
+) {
+    let id_token = jsonwebtoken::encode(
+        &jsonwebtoken::Header::default(),
+        &serde_json::json!({"https://api.openai.com/auth": claims}),
+        &jsonwebtoken::EncodingKey::from_secret(b"executor-auth-test"),
+    )
+    .unwrap();
+    std::fs::write(
+        codex_home.join("auth.json"),
+        serde_json::to_vec(&serde_json::json!({
+            "auth_mode": "chatgpt",
+            "tokens": {
+                "id_token": id_token,
+                "access_token": token,
+                "refresh_token": "test-refresh-token",
+                "account_id": account,
+            },
+            "last_refresh": chrono::Utc::now(),
+        }))
+        .unwrap(),
+    )
+    .unwrap();
+}
+
+async fn stored_chatgpt_provider(codex_home: &Path) -> SharedAuthProvider {
+    let auth_manager = Arc::new(
+        AuthManager::new(
+            codex_home.to_path_buf(),
+            /*enable_codex_api_key_env*/ false,
+            AuthCredentialsStoreMode::File,
+            /*forced_chatgpt_workspace_id*/ None,
+            /*chatgpt_base_url*/ None,
+            AuthKeyringBackendKind::default(),
+            codex_login::test_support::transport_default_auth_route_config(),
+        )
+        .await,
+    );
+    let initial_auth = auth_manager.auth_cached().expect("load initial auth");
+    chatgpt_auth_provider(auth_manager, &initial_auth).unwrap()
+}
+
+#[tokio::test]
+async fn chatgpt_executor_follows_disk_rotation_without_restart() {
+    let home = TempDir::new().unwrap();
+    write_chatgpt_auth(home.path(), "initial-token", "account");
+    let provider = stored_chatgpt_provider(home.path()).await;
+    for token in ["initial-token", "rotated-token", "rotated-again"] {
+        write_chatgpt_auth(home.path(), token, "account");
+        let headers = provider.apply_auth_headers(HeaderMap::new()).await.unwrap();
+        let mut expected = HeaderMap::new();
+        expected.insert(
+            http::header::AUTHORIZATION,
+            HeaderValue::from_str(&format!("Bearer {token}")).unwrap(),
+        );
+        expected.insert("ChatGPT-Account-ID", HeaderValue::from_static("account"));
+        assert_eq!(headers, expected);
+    }
+}
+
+#[tokio::test]
+async fn chatgpt_executor_rejects_account_switch_and_recovers_original_identity() {
+    let home = TempDir::new().unwrap();
+    write_chatgpt_auth(home.path(), "initial-token", "account");
+    let provider = stored_chatgpt_provider(home.path()).await;
+
+    let rejected = provider.to_auth_headers().unwrap();
+    write_chatgpt_auth(home.path(), "other-token", "other-account");
+    assert!(!provider.refresh_after_unauthorized(rejected.clone()).await);
+    assert!(provider.apply_auth_headers(HeaderMap::new()).await.is_err());
+    assert!(provider.to_auth_headers().is_err());
+
+    write_chatgpt_auth(home.path(), "restored-token", "account");
+    assert!(provider.refresh_after_unauthorized(rejected).await);
+    assert_eq!(
+        provider.apply_auth_headers(HeaderMap::new()).await.unwrap()[http::header::AUTHORIZATION],
+        "Bearer restored-token"
+    );
+}
+
+#[tokio::test]
+async fn chatgpt_executor_does_not_reuse_credentials_after_logout() {
+    let home = TempDir::new().unwrap();
+    write_chatgpt_auth(home.path(), "initial-token", "account");
+    let provider = stored_chatgpt_provider(home.path()).await;
+    std::fs::remove_file(home.path().join("auth.json")).unwrap();
+    assert!(provider.apply_auth_headers(HeaderMap::new()).await.is_err());
+    assert!(provider.to_auth_headers().is_err());
+}
+
+#[tokio::test]
+async fn executor_rejects_incomplete_or_empty_bearer() {
+    for missing_refresh in [true, false] {
+        let home = TempDir::new().unwrap();
+        write_chatgpt_auth(home.path(), "initial", "account");
+        let provider = stored_chatgpt_provider(home.path()).await;
+        let path = home.path().join("auth.json");
+        let mut doc: serde_json::Value =
+            serde_json::from_slice(&std::fs::read(&path).unwrap()).unwrap();
+        if missing_refresh {
+            doc.as_object_mut().unwrap().remove("last_refresh");
+        } else {
+            doc["tokens"]["access_token"] = serde_json::json!("");
+        }
+        std::fs::write(path, serde_json::to_vec(&doc).unwrap()).unwrap();
+        assert!(matches!(
+            provider.apply_auth_headers(HeaderMap::new()).await,
+            Err(AuthError::Build(_))
+        ));
+    }
+}
+
+#[tokio::test]
+async fn chatgpt_executor_rejects_rotation_with_incomplete_identity() {
+    for (account, user) in [
+        (Some("account"), None),
+        (None, Some("user")),
+        (Some("account"), Some("")),
+        (Some(" "), Some("user")),
+    ] {
+        let home = TempDir::new().unwrap();
+        let claims = serde_json::json!({"chatgpt_user_id": user});
+        write_chatgpt_auth_with_claims(home.path(), "initial", account, claims.clone());
+        let provider = stored_chatgpt_provider(home.path()).await;
+        assert!(provider.apply_auth_headers(HeaderMap::new()).await.is_ok());
+        assert!(
+            !provider
+                .refresh_after_unauthorized(provider.to_auth_headers().unwrap())
+                .await
+        );
+        write_chatgpt_auth_with_claims(home.path(), "replacement", account, claims);
+        assert!(provider.apply_auth_headers(HeaderMap::new()).await.is_err());
+        assert!(provider.to_auth_headers().is_err());
+    }
+}
+
+#[tokio::test]
+async fn chatgpt_executor_rejects_changed_user_and_workspace_policy() {
+    for claims in [
+        serde_json::json!({"chatgpt_user_id": "other-user", "chatgpt_plan_type": "pro"}),
+        serde_json::json!({"chatgpt_user_id": "user", "chatgpt_plan_type": "enterprise"}),
+        serde_json::json!({"chatgpt_user_id": "user", "chatgpt_plan_type": "pro", "chatgpt_account_is_fedramp": true}),
+    ] {
+        let home = TempDir::new().unwrap();
+        write_chatgpt_auth(home.path(), "initial", "account");
+        let provider = stored_chatgpt_provider(home.path()).await;
+        write_chatgpt_auth_with_claims(home.path(), "replacement", Some("account"), claims);
+        assert!(provider.apply_auth_headers(HeaderMap::new()).await.is_err());
+        assert!(provider.to_auth_headers().is_err());
+    }
+}
+
+#[tokio::test]
+async fn chatgpt_executor_rejects_same_owner_agent_identity() {
+    let home = TempDir::new().unwrap();
+    write_chatgpt_auth_with_claims(
+        home.path(),
+        "initial",
+        Some("test-account-id"),
+        serde_json::json!({"chatgpt_user_id": "test-user-id", "chatgpt_plan_type": "pro"}),
+    );
+    let provider = stored_chatgpt_provider(home.path()).await;
+    let agent_manager = codex_login::test_support::auth_manager_with_agent_identity()
+        .await
+        .unwrap();
+    let Some(CodexAuth::AgentIdentity(agent)) = agent_manager.auth_cached() else {
+        panic!("expected agent identity fixture");
+    };
+    std::fs::write(
+        home.path().join("auth.json"),
+        serde_json::to_vec(&serde_json::json!({
+            "auth_mode": codex_protocol::auth::AuthMode::AgentIdentity,
+            "agent_identity": agent.record(),
+        }))
+        .unwrap(),
+    )
+    .unwrap();
+    // Ensure the fixture is a valid, loadable credential for the same owner.
+    assert!(matches!(
+        CodexAuth::from_auth_storage(
+            home.path(),
+            AuthCredentialsStoreMode::File,
+            /*chatgpt_base_url*/ None,
+            AuthKeyringBackendKind::default(),
+            &codex_login::test_support::transport_default_auth_route_config(),
+        )
+        .await
+        .unwrap(),
+        Some(CodexAuth::AgentIdentity(_))
+    ));
+    assert!(provider.apply_auth_headers(HeaderMap::new()).await.is_err());
+    assert!(provider.to_auth_headers().is_err());
+}
+
+#[tokio::test]
+async fn executor_concurrent_rotation_checks_identity_before_refresh() {
+    const CHILD_MODE: &str = "CODEX_EXECUTOR_AUTH_TEST_MODE";
+    let Ok(mode) = std::env::var(CHILD_MODE) else {
+        // Run in subprocesses so endpoint overrides cannot affect parallel tests.
+        let server = MockServer::start().await;
+        for token in [
+            "at-initial",
+            "at-rotated",
+            "at-retry-original",
+            "at-retry-new",
+        ] {
+            Mock::given(method("GET"))
+                .and(path("/v1/user-auth-credential/whoami"))
+                .and(header("authorization", format!("Bearer {token}")))
+                .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
+                    "chatgpt_user_id": "user",
+                    "chatgpt_account_id": "account",
+                    "chatgpt_plan_type": "pro",
+                    "chatgpt_account_is_fedramp": false,
+                })))
+                .with_priority(2)
+                .up_to_n_times(1)
+                .expect(1)
+                .mount(&server)
+                .await;
+        }
+        Mock::given(method("GET"))
+            .and(path("/v1/user-auth-credential/whoami"))
+            .and(header("authorization", "Bearer at-retry-new"))
+            .respond_with(ResponseTemplate::new(503))
+            .with_priority(1)
+            .up_to_n_times(1)
+            .expect(1)
+            .mount(&server)
+            .await;
+        Mock::given(method("POST"))
+            .and(path("/oauth/token"))
+            .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
+                "access_token": "refreshed",
+                "refresh_token": "refreshed-refresh-token",
+            })))
+            .with_priority(1)
+            .up_to_n_times(1)
+            .expect(1)
+            .mount(&server)
+            .await;
+        Mock::given(wiremock::matchers::any())
+            .respond_with(ResponseTemplate::new(503))
+            .with_priority(10)
+            .expect(0)
+            .mount(&server)
+            .await;
+        for mode in ["pat", "chatgpt", "agent", "kind-switch", "pat-outage"] {
+            let mut child = tokio::process::Command::new(std::env::current_exe().unwrap());
+            child
+                .args([
+                    "--exact",
+                    "exec_server_auth::tests::executor_concurrent_rotation_checks_identity_before_refresh",
+                    "--nocapture",
+                ])
+                .env(CHILD_MODE, mode)
+                .env_remove("CODEX_ACCESS_TOKEN")
+                .env("NO_PROXY", "127.0.0.1,localhost")
+                .env("no_proxy", "127.0.0.1,localhost")
+                .env("CODEX_AUTHAPI_BASE_URL", server.uri())
+                .env("CODEX_AGENT_IDENTITY_AUTHAPI_BASE_URL", server.uri())
+                .env("CODEX_REFRESH_TOKEN_URL_OVERRIDE", format!("{}/oauth/token", server.uri()))
+                .kill_on_drop(true);
+            let output = tokio::time::timeout(Duration::from_secs(30), child.output())
+                .await
+                .unwrap()
+                .unwrap();
+            assert!(output.status.success(), "{output:?}");
+            assert!(String::from_utf8_lossy(&output.stdout).contains("1 passed"));
+        }
+        server.verify().await;
+        return;
+    };
+
+    let home = TempDir::new().unwrap();
+    let auth_path = home.path().join("auth.json");
+    if mode == "pat" || mode == "pat-outage" {
+        let token = if mode == "pat" {
+            "at-initial"
+        } else {
+            "at-retry-original"
+        };
+        std::fs::write(
+            &auth_path,
+            serde_json::to_vec(&serde_json::json!({"personal_access_token":token})).unwrap(),
+        )
+        .unwrap();
+    } else {
+        write_chatgpt_auth(home.path(), "initial", "account");
+    }
+    let provider = stored_chatgpt_provider(home.path()).await;
+    if mode == "pat-outage" {
+        std::fs::write(&auth_path, r#"{"personal_access_token":"at-retry-new"}"#).unwrap();
+        assert!(matches!(
+            provider.apply_auth_headers(HeaderMap::new()).await,
+            Err(AuthError::Transient(_))
+        ));
+        assert!(provider.to_auth_headers().is_err());
+        let headers = provider.apply_auth_headers(HeaderMap::new()).await.unwrap();
+        assert_eq!(headers[http::header::AUTHORIZATION], "Bearer at-retry-new");
+        assert_eq!(provider.to_auth_headers().unwrap(), headers);
+        return;
+    }
+    if mode == "agent" || mode == "kind-switch" {
+        if mode == "agent" {
+            let manager = codex_login::test_support::auth_manager_with_agent_identity()
+                .await
+                .unwrap();
+            let Some(CodexAuth::AgentIdentity(agent)) = manager.auth_cached() else {
+                panic!("agent fixture");
+            };
+            let mut record = agent.record().clone();
+            record.task_id = None;
+            std::fs::write(&auth_path, serde_json::to_vec(&serde_json::json!({"auth_mode":codex_protocol::auth::AuthMode::AgentIdentity, "agent_identity":record})).unwrap()).unwrap();
+        } else {
+            std::fs::write(&auth_path, r#"{"personal_access_token":"at-initial"}"#).unwrap();
+        }
+        assert!(matches!(
+            provider.apply_auth_headers(HeaderMap::new()).await,
+            Err(AuthError::Build(_))
+        ));
+        return;
+    }
+    let expected_bearer = if mode == "pat" {
+        std::fs::write(&auth_path, r#"{"personal_access_token":"at-rotated"}"#).unwrap();
+        "Bearer at-rotated"
+    } else {
+        let expired = jsonwebtoken::encode(
+            &jsonwebtoken::Header::default(),
+            &serde_json::json!({"exp": chrono::Utc::now().timestamp() - 60}),
+            &jsonwebtoken::EncodingKey::from_secret(b"executor-auth-test"),
+        )
+        .unwrap();
+        for (account, user) in [
+            ("other-account", "user"),
+            ("account", "other-user"),
+            ("account", "user"),
+        ] {
+            write_chatgpt_auth_with_claims(
+                home.path(),
+                &expired,
+                Some(account),
+                serde_json::json!({"chatgpt_user_id": user, "chatgpt_plan_type": "pro"}),
+            );
+            let original = std::fs::read(&auth_path).unwrap();
+            if account != "account" || user != "user" {
+                assert!(provider.apply_auth_headers(HeaderMap::new()).await.is_err());
+                assert_eq!(std::fs::read(&auth_path).unwrap(), original);
+            }
+        }
+        "Bearer refreshed"
+    };
+    let barrier = Arc::new(tokio::sync::Barrier::new(32));
+    let results = futures::future::join_all((0..32).map(|_| {
+        let provider = Arc::clone(&provider);
+        let barrier = Arc::clone(&barrier);
+        async move {
+            barrier.wait().await;
+            provider.apply_auth_headers(HeaderMap::new()).await
+        }
+    }))
+    .await;
+    for result in results {
+        assert_eq!(
+            result.unwrap()[http::header::AUTHORIZATION],
+            expected_bearer
+        );
+    }
+}
+
+#[tokio::test]
+async fn executor_refresh_survives_cancelled_caller() {
+    use std::process::Stdio;
+    use tokio::io::AsyncBufReadExt;
+    use tokio::io::AsyncReadExt;
+    use tokio::io::AsyncWriteExt;
+
+    const CHILD_MODE: &str = "CODEX_EXECUTOR_CANCEL_TEST_MODE";
+    let Ok(mode) = std::env::var(CHILD_MODE) else {
+        // Isolate endpoint overrides; release the response only after the caller
+        // is cancelled, with no sleep or race against the HTTP response.
+        for mode in ["proactive", "recovery", "logout", "replacement", "timeout"] {
+            tokio::time::timeout(Duration::from_secs(30), async {
+                let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+                let endpoint = format!("http://{}/token", listener.local_addr().unwrap());
+                let mut child = tokio::process::Command::new(std::env::current_exe().unwrap())
+                    .args(["--exact", "exec_server_auth::tests::executor_refresh_survives_cancelled_caller", "--nocapture"])
+                    .env(CHILD_MODE, mode)
+                    .env_remove("CODEX_ACCESS_TOKEN")
+                    .env("CODEX_REFRESH_TOKEN_URL_OVERRIDE", endpoint)
+                    .env("NO_PROXY", "127.0.0.1,localhost")
+                    .env("no_proxy", "127.0.0.1,localhost")
+                    .stdin(Stdio::piped())
+                    .stdout(Stdio::piped())
+                    .stderr(Stdio::piped())
+                    .kill_on_drop(true)
+                    .spawn().unwrap();
+                let (mut stream, _) = listener.accept().await.unwrap();
+                let mut request = tokio::io::BufReader::new(&mut stream);
+                let mut content_length = 0;
+                loop {
+                    let mut line = String::new();
+                    assert!(request.read_line(&mut line).await.unwrap() > 0);
+                    if line == "\r\n" { break; }
+                    if let Some(length) = line.to_ascii_lowercase().strip_prefix("content-length:") {
+                        content_length = length.trim().parse().unwrap();
+                    }
+                }
+                request.read_exact(&mut vec![0; content_length]).await.unwrap();
+                drop(request);
+                child.stdin.take().unwrap().write_all(b"cancel\n").await.unwrap();
+                let mut stdout = tokio::io::BufReader::new(child.stdout.take().unwrap());
+                loop {
+                    let mut line = String::new();
+                    assert!(stdout.read_line(&mut line).await.unwrap() > 0);
+                    if line.contains("CALLER_CANCELLED") { break; }
+                }
+                let body = r#"{"access_token":"rotated-access","refresh_token":"rotated-refresh"}"#;
+                if mode != "timeout" {
+                    stream.write_all(format!(
+                        "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", body.len()
+                    ).as_bytes()).await.unwrap();
+                }
+                // Keep the unanswered timeout connection open until the child finishes.
+                let mut remaining = String::new();
+                stdout.read_to_string(&mut remaining).await.unwrap();
+                let output = child.wait_with_output().await.unwrap();
+                assert!(output.status.success(), "{remaining}\n{output:?}");
+                assert!(remaining.contains("1 passed"), "{remaining}");
+            }).await.expect("cancelled refresh must finish persisting");
+        }
+        return;
+    };
+
+    let home = TempDir::new().unwrap();
+    let token = if mode != "recovery" {
+        jsonwebtoken::encode(
+            &jsonwebtoken::Header::default(),
+            &serde_json::json!({"exp": chrono::Utc::now().timestamp() - 60}),
+            &jsonwebtoken::EncodingKey::from_secret(b"executor-auth-test"),
+        )
+        .unwrap()
+    } else {
+        "initial-access".to_string()
+    };
+    write_chatgpt_auth(home.path(), &token, "account");
+    let provider = stored_chatgpt_provider(home.path()).await;
+    let caller = Arc::clone(&provider);
+    let rejected = provider.to_auth_headers().unwrap();
+    let refresh_mode = mode.clone();
+    let task = tokio::spawn(async move {
+        if refresh_mode != "recovery" {
+            caller.apply_auth_headers(HeaderMap::new()).await.unwrap();
+        } else {
+            assert!(caller.refresh_after_unauthorized(rejected).await);
+        }
+    });
+    let mut instruction = String::new();
+    tokio::io::BufReader::new(tokio::io::stdin())
+        .read_line(&mut instruction)
+        .await
+        .unwrap();
+    assert_eq!(instruction, "cancel\n");
+    task.abort();
+    assert!(task.await.unwrap_err().is_cancelled());
+    if mode == "logout" {
+        std::fs::remove_file(home.path().join("auth.json")).unwrap();
+    } else if mode == "replacement" {
+        write_chatgpt_auth(home.path(), "fresh-external-access", "account");
+        let path = home.path().join("auth.json");
+        let mut stored: serde_json::Value =
+            serde_json::from_slice(&std::fs::read(&path).unwrap()).unwrap();
+        stored["tokens"]["refresh_token"] = serde_json::json!("externally-rotated-refresh");
+        std::fs::write(path, serde_json::to_vec(&stored).unwrap()).unwrap();
+    }
+    println!("CALLER_CANCELLED");
+    if mode == "timeout" {
+        tokio::time::pause();
+        tokio::time::advance(Duration::from_secs(31)).await;
+        tokio::time::resume();
+        assert!(matches!(
+            provider.apply_auth_headers(HeaderMap::new()).await,
+            Err(AuthError::Transient(_))
+        ));
+        // Repair is usable without releasing the old response or restarting the executor.
+        write_chatgpt_auth(home.path(), "fresh-external-access", "account");
+    }
+    if mode == "logout" {
+        assert!(matches!(
+            provider.apply_auth_headers(HeaderMap::new()).await,
+            Err(AuthError::Build(_))
+        ));
+        assert!(!home.path().join("auth.json").exists());
+        return;
+    }
+    let repaired = mode == "replacement" || mode == "timeout";
+    // This waits on the same permit, proving the detached operation kept it
+    // until both persistence and cache update were complete.
+    let headers = provider.apply_auth_headers(HeaderMap::new()).await.unwrap();
+    assert_eq!(
+        headers[http::header::AUTHORIZATION],
+        if repaired {
+            "Bearer fresh-external-access"
+        } else {
+            "Bearer rotated-access"
+        }
+    );
+    let stored: serde_json::Value =
+        serde_json::from_slice(&std::fs::read(home.path().join("auth.json")).unwrap()).unwrap();
+    assert_eq!(
+        stored["tokens"]["access_token"],
+        if repaired {
+            "fresh-external-access"
+        } else {
+            "rotated-access"
+        }
+    );
+    assert_eq!(
+        stored["tokens"]["refresh_token"],
+        match mode.as_str() {
+            "replacement" => "externally-rotated-refresh",
+            "timeout" => "test-refresh-token",
+            _ => "rotated-refresh",
+        }
+    );
+}
+
 async fn test_provider() -> AwsSigV4AuthProvider {
     let context = AwsAuthContext::load_with_access_keys(
         AwsAuthConfig {