Guard auth persistence and reloads against concurrent login changes (#55646)

## Why

An in-flight token refresh or auth reload can outlive a login change or logout. Its result must not overwrite credentials for a different login or restore stale cached auth.

## What changed

- Serialize in-process auth storage reads and writes that participate in credential updates.
- Verify the stored login before persisting refreshed tokens or agent identity records, while allowing older clients to drop login metadata when the refresh token still matches.
- Reject stale auth reload results using generation checks, and invalidate pending reads on logout even when the cache is already empty.
- Keep account grant reads strict when the keyring is unavailable, while preserving ordinary auth fallback behavior.

## Testing

Add regression coverage for stale reloads after logout, login replacement during refresh, legacy metadata loss, keyring recovery with rotated tokens, and refresh persistence failures.

GitOrigin-RevId: 01bce9ddbc3101de868adcda3370fcdc4eb70163

codex-rs/login/src/auth/account_grant.rs

@@ -80,8 +80,11 @@ impl AccountOAuthGrant {
 }
 
 pub(super) fn same_account_grant(a: &TokenData, b: &TokenData) -> bool {
-    a.oauth_provenance == b.oauth_provenance
-        && a.login_owner_id == b.login_owner_id
+    a.oauth_provenance == b.oauth_provenance && same_account_login(a, b)
+}
+
+pub(super) fn same_account_login(a: &TokenData, b: &TokenData) -> bool {
+    a.login_owner_id == b.login_owner_id
         && a.account_id == b.account_id
         && a.id_token.chatgpt_user_id == b.id_token.chatgpt_user_id
 }

codex-rs/login/src/auth/auth_tests.rs

@@ -80,9 +80,14 @@ async fn refresh_without_id_token() -> anyhow::Result<()> {
         issuer: "https://idp.example".to_string(),
         client_id: "codex-client".to_string(),
     };
+    let initial = storage
+        .load()?
+        .context("stored auth")?
+        .tokens
+        .context("stored tokens")?;
     let updated = super::persist_tokens(
         &storage,
-        "test-refresh-token",
+        &initial,
         Some(provenance.clone()),
         /*id_token*/ None,
         Some("new-access-token".to_string()),
@@ -97,30 +102,35 @@ async fn refresh_without_id_token() -> anyhow::Result<()> {
 
     let updated = super::persist_tokens(
         &storage,
-        "new-refresh-token",
-        Some(provenance.clone()),
+        &tokens,
+        /*provenance*/ None,
         /*id_token*/ None,
         Some("later-access-token".to_string()),
         /*refresh_token*/ None,
     )?;
-    let tokens = updated.tokens.context("persisted auth is missing tokens")?;
-    assert_eq!(tokens.refresh_token, "new-refresh-token");
-    assert_eq!(tokens.oauth_provenance, Some(provenance.clone()));
-
-    // A response without a rotated token cannot establish authority for a token that
-    // another writer installed after this refresh began.
-    let updated = super::persist_tokens(
-        &storage,
-        "superseded-refresh-token",
-        Some(provenance),
-        /*id_token*/ None,
-        Some("latest-access-token".to_string()),
-        /*refresh_token*/ None,
-    )?;
-    assert_eq!(storage.load()?, Some(updated.clone()));
+    let previous = tokens;
     let tokens = updated.tokens.context("persisted auth is missing tokens")?;
     assert_eq!(tokens.refresh_token, "new-refresh-token");
     assert_eq!(tokens.oauth_provenance, None);
+
+    // Preserve ordinary same-login refresh behavior when the stored tokens changed.
+    let mut expected = previous.clone();
+    expected.access_token = "stale-access-token".into();
+    expected.refresh_token = "stale-refresh-token".into();
+    expected.oauth_provenance = Some(provenance.clone());
+    assert_eq!(
+        super::persist_tokens(
+            &storage,
+            &previous,
+            Some(provenance),
+            /*id_token*/ None,
+            Some("stale-access-token".into()),
+            Some("stale-refresh-token".into()),
+        )?
+        .tokens,
+        Some(expected)
+    );
+
     Ok(())
 }
 
@@ -600,6 +610,10 @@ async fn chatgpt_auth_registers_agent_identity_when_enabled() -> anyhow::Result<
         },
         codex_home.path(),
     )?;
+    let storage = FileAuthStorage::new(codex_home.path().to_path_buf());
+    let mut stored = storage.load()?.expect("auth should exist");
+    stored.tokens.as_mut().unwrap().login_owner_id = Some("login-a".into());
+    storage.save(&stored)?;
     let auth = super::load_auth(
         codex_home.path(),
         /*enable_codex_api_key_env*/ false,
@@ -614,6 +628,10 @@ async fn chatgpt_auth_registers_agent_identity_when_enabled() -> anyhow::Result<
     .await?
     .expect("auth should load");
 
+    // An older client rewrites the same credentials without login metadata.
+    stored.tokens.as_mut().unwrap().login_owner_id = None;
+    storage.save(&stored)?;
+
     assert!(
         auth.agent_identity_auth(
             AgentIdentityAuthPolicy::JwtOnly,
@@ -682,6 +700,8 @@ async fn chatgpt_auth_registers_agent_identity_when_enabled() -> anyhow::Result<
         .expect("identity should persist");
     assert_eq!(persisted.agent_runtime_id, "agent-runtime-123");
     assert_eq!(persisted.task_id.as_deref(), Some("task-123"));
+    stored.agent_identity = Some(AgentIdentityStorage::Record(agent_auth.record().clone()));
+    assert_eq!(storage.load()?, Some(stored));
 
     let reloaded = super::load_auth(
         codex_home.path(),

codex-rs/login/src/auth/change_state.rs

@@ -5,7 +5,7 @@ use super::CodexAuth;
 /// Opaque revisions local to one auth manager. Consumers must reset on reconnect.
 #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
 pub struct AuthChangeState {
-    /// Advances whenever cached credentials change.
+    /// Advances on cached credential changes and explicit invalidation of pending reads.
     pub generation: u64,
     /// Advances on login, logout, or changes of user, workspace, or auth mode.
     /// A verified legacy-to-OAuth credential migration preserves the same owner.

codex-rs/login/src/auth/change_state_tests.rs

@@ -74,12 +74,27 @@ async fn auth_owner_generation_preserves_logout_and_switches_when_coalesced() {
         AuthManager::from_auth_for_testing_with_home(initial.clone(), home.path().to_path_buf());
     let mut changes = manager.auth_change_state_receiver();
 
+    let generation = changes.borrow().generation;
     manager.logout().await.unwrap();
+    assert!(
+        manager
+            .publish_reloaded_auth(generation, Some(initial.clone()))
+            .is_none()
+    );
+    assert!(manager.auth_cached().is_none());
+    // Deletion also fences a read when the cache was already empty.
+    let generation = changes.borrow().generation;
+    manager.logout().await.unwrap();
+    assert!(
+        manager
+            .publish_reloaded_auth(generation, Some(initial.clone()))
+            .is_none()
+    );
     manager.set_cached_auth(Some(initial.clone()));
     assert_eq!(
         *changes.borrow_and_update(),
         AuthChangeState {
-            generation: 2,
+            generation: 4,
             owner_generation: 2,
             source_generation: 2,
             login_generation: 2,
@@ -91,7 +106,7 @@ async fn auth_owner_generation_preserves_logout_and_switches_when_coalesced() {
     assert_eq!(
         *changes.borrow_and_update(),
         AuthChangeState {
-            generation: 4,
+            generation: 6,
             owner_generation: 4,
             source_generation: 4,
             login_generation: 4,

codex-rs/login/src/auth/manager.rs

@@ -5,6 +5,7 @@ mod account_grant;
 pub use account_grant::AccountOAuthGrant;
 pub use account_grant::AccountOAuthGrantLease;
 use account_grant::same_account_grant;
+use account_grant::same_account_login;
 
 use chrono::Utc;
 use http::StatusCode;
@@ -63,6 +64,7 @@ pub use crate::auth::storage::AgentIdentityStorage;
 pub use crate::auth::storage::AuthDotJson;
 pub use crate::auth::storage::AuthKeyringBackendKind;
 use crate::auth::storage::AuthStorageBackend;
+use crate::auth::storage::auth_storage_commit_guard;
 use crate::auth::storage::create_auth_storage;
 use crate::default_client::create_client;
 use crate::default_client::create_default_auth_client;
@@ -952,11 +954,23 @@ fn persist_agent_identity_record(
     storage: &Arc<dyn AuthStorageBackend>,
     record: AgentIdentityAuthRecord,
 ) -> std::io::Result<()> {
+    let _commit = auth_storage_commit_guard()?;
     let mut guard = auth_dot_json
         .lock()
         .map_err(|_| std::io::Error::other("failed to lock auth state"))?;
-    let mut auth = storage
-        .load()?
+    let stored = storage.load()?;
+    if let Some(previous) = guard.as_ref().and_then(|auth| auth.tokens.as_ref())
+        && previous.login_owner_id.is_some()
+        && !stored
+            .as_ref()
+            .and_then(|auth| auth.tokens.as_ref())
+            .is_some_and(|current| same_login_for_persistence(previous, current))
+    {
+        return Err(std::io::Error::other(
+            "Account changed during agent identity registration.",
+        ));
+    }
+    let mut auth = stored
         .or_else(|| guard.clone())
         .ok_or_else(|| std::io::Error::other("auth data is not available"))?;
     auth.agent_identity = Some(AgentIdentityStorage::Record(record));
@@ -1002,6 +1016,7 @@ pub fn logout(
     auth_credentials_store_mode: AuthCredentialsStoreMode,
     keyring_backend_kind: AuthKeyringBackendKind,
 ) -> std::io::Result<bool> {
+    let _commit = auth_storage_commit_guard()?;
     let storage = create_auth_storage(
         codex_home.to_path_buf(),
         auth_credentials_store_mode,
@@ -1172,6 +1187,7 @@ pub fn save_auth(
     auth_credentials_store_mode: AuthCredentialsStoreMode,
     keyring_backend_kind: AuthKeyringBackendKind,
 ) -> std::io::Result<()> {
+    let _commit = auth_storage_commit_guard()?;
     let storage = create_auth_storage(
         codex_home.to_path_buf(),
         auth_credentials_store_mode,
@@ -1190,6 +1206,7 @@ pub fn load_auth_dot_json(
     auth_credentials_store_mode: AuthCredentialsStoreMode,
     keyring_backend_kind: AuthKeyringBackendKind,
 ) -> std::io::Result<Option<AuthDotJson>> {
+    let _commit = auth_storage_commit_guard()?;
     let storage = create_auth_storage(
         codex_home.to_path_buf(),
         auth_credentials_store_mode,
@@ -1615,24 +1632,49 @@ async fn load_auth(
     Ok(Some(auth))
 }
 
+// Ordinary auth tolerates legacy metadata loss; EMA grant checks remain strict.
+fn same_login_for_persistence(expected: &TokenData, stored: &TokenData) -> bool {
+    same_account_login(expected, stored)
+        || (expected.login_owner_id.is_some()
+            && stored.login_owner_id.is_none()
+            && stored.account_id == expected.account_id
+            && stored.id_token.chatgpt_user_id == expected.id_token.chatgpt_user_id
+            && !stored.refresh_token.is_empty()
+            && stored.refresh_token == expected.refresh_token)
+}
+
 // Persist refreshed tokens into auth storage and update last_refresh.
-fn persist_tokens(
+pub(super) fn persist_tokens(
     storage: &Arc<dyn AuthStorageBackend>,
-    previous_refresh_token: &str,
+    expected: &TokenData,
     provenance: Option<AccountOAuthProvenance>,
     id_token: Option<String>,
     access_token: Option<String>,
     refresh_token: Option<String>,
 ) -> std::io::Result<AuthDotJson> {
-    let mut auth_dot_json = storage
-        .load()?
+    let _commit = auth_storage_commit_guard()?;
+    let stored = storage.load()?;
+    let mut auth_dot_json = stored.ok_or(std::io::Error::other("Token data is not available."))?;
+    if auth_dot_json.resolved_mode() != AuthMode::Chatgpt {
+        return Err(std::io::Error::other(
+            "Account credentials changed during refresh.",
+        ));
+    }
+    if !auth_dot_json
+        .tokens
+        .as_ref()
+        .is_some_and(|tokens| same_login_for_persistence(expected, tokens))
+    {
+        return Err(std::io::Error::other(
+            "Account credentials changed during refresh.",
+        ));
+    }
+    let tokens = auth_dot_json
+        .tokens
+        .as_mut()
         .ok_or(std::io::Error::other("Token data is not available."))?;
-
-    let tokens = auth_dot_json.tokens.get_or_insert_with(TokenData::default);
-    // If a concurrent writer replaced the stored refresh token and the response did not
-    // rotate it, its authority is unknown. Do not attach the earlier grant's provenance.
     tokens.oauth_provenance = provenance
-        .filter(|_| refresh_token.is_some() || tokens.refresh_token == previous_refresh_token);
+        .filter(|_| refresh_token.is_some() || tokens.refresh_token == expected.refresh_token);
     if let Some(id_token) = id_token {
         tokens.id_token = parse_chatgpt_jwt_claims(&id_token).map_err(std::io::Error::other)?;
     }
@@ -1820,6 +1862,11 @@ impl AuthDotJson {
     }
 }
 
+struct AuthReload {
+    changed: bool,
+    changed_for_refresh: bool,
+}
+
 /// Internal cached auth state.
 #[derive(Clone)]
 struct CachedAuth {
@@ -2563,19 +2610,12 @@ impl AuthManager {
         let home = self.codex_home.clone();
         let mode = self.auth_credentials_store_mode;
         let backend = self.keyring_backend_kind;
+        let originator = codex_otel::auth_storage::AuthStorageOriginator::current();
         let stored = tokio::task::spawn_blocking(move || {
-            // Unlike ordinary login loading, an unavailable keyring must not authorize
-            // account exchange using an older file fallback.
-            if mode == AuthCredentialsStoreMode::Auto {
-                let keyring =
-                    create_auth_storage(home.clone(), AuthCredentialsStoreMode::Keyring, backend);
-                if let Some(auth) = keyring.load()? {
-                    return Ok(Some(auth));
-                }
-                create_auth_storage(home, AuthCredentialsStoreMode::File, backend).load()
-            } else {
-                create_auth_storage(home, mode, backend).load()
-            }
+            originator.sync_scope(|| {
+                let _commit = auth_storage_commit_guard()?;
+                create_auth_storage(home, mode, backend).load_account_grant()
+            })
         })
         .await
         .map_err(|_| std::io::Error::other("Account credential read failed."))?
@@ -2606,8 +2646,16 @@ impl AuthManager {
     /// Reloads auth from the active source. Returns whether the auth value changed.
     pub async fn reload(&self) -> bool {
         tracing::info!("Reloading auth");
-        let new_auth = self.load_auth().await;
-        self.set_cached_auth(new_auth)
+        loop {
+            let generation = self.auth_change_state_tx.borrow().generation;
+            let new_auth = self.load_auth().await;
+            if let Some(publication) = self.publish_reloaded_auth(generation, new_auth) {
+                return publication.changed;
+            }
+            if self.inner.is_poisoned() {
+                return false;
+            }
+        }
     }
 
     async fn reload_if_account_id_matches(
@@ -2622,26 +2670,30 @@ impl AuthManager {
             }
         };
 
-        let new_auth = self.load_auth().await;
-        let new_account_id = new_auth.as_ref().and_then(CodexAuth::get_account_id);
+        loop {
+            let generation = self.auth_change_state_tx.borrow().generation;
+            let new_auth = self.load_auth().await;
+            let new_account_id = new_auth.as_ref().and_then(CodexAuth::get_account_id);
 
-        if new_account_id.as_deref() != Some(expected_account_id) {
-            let found_account_id = new_account_id.as_deref().unwrap_or("unknown");
-            tracing::info!(
-                "Skipping auth reload due to account id mismatch (expected: {expected_account_id}, found: {found_account_id})"
-            );
-            return ReloadOutcome::Skipped;
-        }
+            if new_account_id.as_deref() != Some(expected_account_id) {
+                let found_account_id = new_account_id.as_deref().unwrap_or("unknown");
+                tracing::info!(
+                    "Skipping auth reload due to account id mismatch (expected: {expected_account_id}, found: {found_account_id})"
+                );
+                return ReloadOutcome::Skipped;
+            }
 
-        tracing::info!("Reloading auth for account {expected_account_id}");
-        let cached_before_reload = self.auth_cached();
-        let auth_changed =
-            !Self::auths_equal_for_refresh(cached_before_reload.as_ref(), new_auth.as_ref());
-        self.set_cached_auth(new_auth);
-        if auth_changed {
-            ReloadOutcome::ReloadedChanged
-        } else {
-            ReloadOutcome::ReloadedNoChange
+            tracing::info!("Reloading auth for account {expected_account_id}");
+            if let Some(publication) = self.publish_reloaded_auth(generation, new_auth) {
+                return if publication.changed_for_refresh {
+                    ReloadOutcome::ReloadedChanged
+                } else {
+                    ReloadOutcome::ReloadedNoChange
+                };
+            }
+            if self.inner.is_poisoned() {
+                return ReloadOutcome::Skipped;
+            }
         }
     }
 
@@ -2753,46 +2805,71 @@ impl AuthManager {
     }
 
     fn set_cached_auth(&self, new_auth: Option<CodexAuth>) -> bool {
-        if let Ok(mut guard) = self.inner.write() {
-            let previous = guard.auth.as_ref();
-            let changed = !AuthManager::auths_equal(previous, new_auth.as_ref());
-            let auth_changed_for_refresh =
-                !Self::auths_equal_for_refresh(previous, new_auth.as_ref());
-            let owner_changed =
-                auth_changed_for_refresh && !same_owner(previous, new_auth.as_ref());
-            let source_changed = auth_changed_for_refresh
-                && !super::source::same_source(previous, new_auth.as_ref());
-            let login_changed =
-                auth_changed_for_refresh && !same_login(previous, new_auth.as_ref());
-            if owner_changed {
-                self.auth_route_config
-                    .application_network_policy()
-                    .invalidate();
-            }
-            if auth_changed_for_refresh {
-                guard.permanent_refresh_failure = None;
-            }
-            tracing::info!("Reloaded auth, changed: {changed}");
-            guard.auth = new_auth;
-            if auth_changed_for_refresh {
-                self.auth_change_state_tx.send_modify(|state| {
-                    state.generation += 1;
-                    if owner_changed {
-                        state.owner_generation += 1;
-                    }
-                    if source_changed {
-                        state.source_generation += 1;
-                    }
-                    if login_changed {
-                        state.login_generation += 1;
-                    }
-                });
-                self.auth_change_tx.send_modify(|revision| *revision += 1);
-            }
-            changed
-        } else {
-            false
+        let Ok(mut guard) = self.inner.write() else {
+            return false;
+        };
+        self.set_cached_auth_locked(&mut guard, new_auth)
+    }
+
+    fn publish_reloaded_auth(
+        &self,
+        generation: u64,
+        auth: Option<CodexAuth>,
+    ) -> Option<AuthReload> {
+        let mut guard = self.inner.write().ok()?;
+        if self.auth_change_state_tx.borrow().generation != generation {
+            return None;
         }
+        let changed_for_refresh =
+            !Self::auths_equal_for_refresh(guard.auth.as_ref(), auth.as_ref());
+        let changed = self.set_cached_auth_locked(&mut guard, auth);
+        Some(AuthReload {
+            changed,
+            changed_for_refresh,
+        })
+    }
+
+    fn invalidate_pending_auth_reads(&self) {
+        if let Ok(_guard) = self.inner.write() {
+            self.auth_change_state_tx
+                .send_modify(|state| state.generation += 1);
+        }
+    }
+
+    fn set_cached_auth_locked(&self, guard: &mut CachedAuth, new_auth: Option<CodexAuth>) -> bool {
+        let previous = guard.auth.as_ref();
+        let changed = !AuthManager::auths_equal(previous, new_auth.as_ref());
+        let auth_changed_for_refresh = !Self::auths_equal_for_refresh(previous, new_auth.as_ref());
+        let owner_changed = auth_changed_for_refresh && !same_owner(previous, new_auth.as_ref());
+        let source_changed =
+            auth_changed_for_refresh && !super::source::same_source(previous, new_auth.as_ref());
+        let login_changed = auth_changed_for_refresh && !same_login(previous, new_auth.as_ref());
+        if owner_changed {
+            self.auth_route_config
+                .application_network_policy()
+                .invalidate();
+        }
+        if auth_changed_for_refresh {
+            guard.permanent_refresh_failure = None;
+        }
+        tracing::info!("Reloaded auth, changed: {changed}");
+        guard.auth = new_auth;
+        if auth_changed_for_refresh {
+            self.auth_change_state_tx.send_modify(|state| {
+                state.generation += 1;
+                if owner_changed {
+                    state.owner_generation += 1;
+                }
+                if source_changed {
+                    state.source_generation += 1;
+                }
+                if login_changed {
+                    state.login_generation += 1;
+                }
+            });
+            self.auth_change_tx.send_modify(|revision| *revision += 1);
+        }
+        changed
     }
 
     pub async fn set_external_auth(
@@ -3108,6 +3185,7 @@ impl AuthManager {
             self.keyring_backend_kind,
         )?;
         // Always reload to clear cached auth, even if no stored credentials were present.
+        self.invalidate_pending_auth_reads();
         self.clear_external_auth();
         self.reload().await;
         Ok(removed)
@@ -3128,6 +3206,7 @@ impl AuthManager {
             self.keyring_backend_kind,
         )?;
         // Always reload to clear cached auth, even if no stored credentials were present.
+        self.invalidate_pending_auth_reads();
         self.clear_external_auth();
         self.reload().await;
         Ok(result)
@@ -3265,13 +3344,13 @@ impl AuthManager {
 
         // An explicit override may send refresh to another authority. Retain provenance
         // only when the endpoint used by this request matches the recorded issuer.
-        let provenance = tokens.oauth_provenance.filter(|provenance| {
+        let provenance = tokens.oauth_provenance.clone().filter(|provenance| {
             endpoint == format!("{}/oauth/token", provenance.issuer.trim_end_matches('/'))
         });
 
         persist_tokens(
             auth.storage(),
-            &tokens.refresh_token,
+            &tokens,
             provenance,
             refresh_response.id_token,
             refresh_response.access_token,

codex-rs/login/src/auth/storage.rs

@@ -172,8 +172,21 @@ pub(super) fn delete_file_if_exists(codex_home: &Path) -> std::io::Result<bool>
     }
 }
 
+// Protect in-process read-modify-write commits, never network requests.
+static AUTH_STORAGE_COMMIT: Mutex<()> = Mutex::new(());
+
+pub(super) fn auth_storage_commit_guard() -> std::io::Result<std::sync::MutexGuard<'static, ()>> {
+    AUTH_STORAGE_COMMIT
+        .lock()
+        .map_err(|_| std::io::Error::other("Auth storage commit lock is unavailable."))
+}
+
 pub(super) trait AuthStorageBackend: Debug + Send + Sync {
     fn load(&self) -> std::io::Result<Option<AuthDotJson>>;
+    /// An unavailable keyring cannot authorize using an older fallback record.
+    fn load_account_grant(&self) -> std::io::Result<Option<AuthDotJson>> {
+        self.load()
+    }
     fn save(&self, auth: &AuthDotJson) -> std::io::Result<()>;
     fn delete(&self) -> std::io::Result<bool>;
 }
@@ -476,6 +489,13 @@ impl AutoAuthStorage {
 }
 
 impl AuthStorageBackend for AutoAuthStorage {
+    fn load_account_grant(&self) -> std::io::Result<Option<AuthDotJson>> {
+        match self.keyring_storage.load()? {
+            Some(auth) => Ok(Some(auth)),
+            None => self.file_storage.load(),
+        }
+    }
+
     fn load(&self) -> std::io::Result<Option<AuthDotJson>> {
         let mut telemetry = storage_telemetry::telemetry(
             AuthCredentialsStoreMode::Auto,

codex-rs/login/src/auth/storage_policy_tests.rs

@@ -298,7 +298,62 @@ fn auto_auth_storage_load_falls_back_when_keyring_errors() -> anyhow::Result<()>
     storage.file_storage.save(&expected)?;
 
     let loaded = storage.load()?;
-    assert_eq!(loaded, Some(expected));
+    assert_eq!(loaded, Some(expected.clone()));
+    mock_keyring.set_error(&key, KeyringError::Invalid("error".into(), "load".into()));
+    assert!(storage.load_account_grant().is_err());
+    assert_eq!(storage.load()?, Some(encrypted));
+    assert_eq!(storage.file_storage.load()?, Some(expected));
+
+    // Preserve rotated tokens whether the keyring recovers before or during persistence.
+    let mut previous = auth_with_prefix("same-login");
+    previous.auth_mode = Some(AuthMode::Chatgpt);
+    previous.openai_api_key = None;
+    let backend: Arc<dyn AuthStorageBackend> = Arc::new(storage.clone());
+    for recover_before_persist in [false, true] {
+        seed_secrets_backend_with_auth(&mock_keyring, codex_home.path(), &previous)?;
+        let mut cached = previous.clone();
+        if recover_before_persist {
+            let tokens = cached.tokens.as_mut().context("fallback account tokens")?;
+            tokens.access_token = "fallback-access".into();
+            tokens.refresh_token = "fallback-refresh".into();
+        }
+        storage.file_storage.save(&cached)?;
+        mock_keyring.set_error(&key, KeyringError::Invalid("error".into(), "load".into()));
+        if recover_before_persist {
+            assert_eq!(storage.load()?, Some(cached.clone()));
+        }
+        let refreshed = crate::auth::manager::persist_tokens(
+            &backend,
+            cached.tokens.as_ref().context("stored account tokens")?,
+            /*provenance*/ None,
+            /*id_token*/ None,
+            Some("rotated-access".into()),
+            Some("rotated-refresh".into()),
+        )?;
+        let tokens = refreshed
+            .tokens
+            .as_ref()
+            .context("refreshed account tokens")?;
+        assert_eq!(tokens.access_token, "rotated-access");
+        assert_eq!(tokens.refresh_token, "rotated-refresh");
+        assert_eq!(storage.load()?, Some(refreshed.clone()));
+        assert_keyring_saved_auth_and_removed_fallback(
+            &mock_keyring,
+            codex_home.path(),
+            &refreshed,
+        )?;
+    }
+
+    // An absent fallback must not select a plaintext destination for cached credentials.
+    storage.file_storage.delete()?;
+    mock_keyring.set_error(&key, KeyringError::Invalid("error".into(), "load".into()));
+    assert_eq!(storage.load()?, None);
+    storage.save(&auth_with_prefix("cached"))?;
+    assert_keyring_saved_auth_and_removed_fallback(
+        &mock_keyring,
+        codex_home.path(),
+        &auth_with_prefix("cached"),
+    )?;
     Ok(())
 }
 

codex-rs/login/tests/suite/auth_refresh.rs

@@ -191,7 +191,7 @@ async fn refresh_token_honors_respect_system_proxy() -> Result<()> {
 
 #[serial_test::serial(auth_env)]
 #[tokio::test]
-async fn refresh_token_succeeds_updates_storage() -> Result<()> {
+async fn account_grant_tracks_refresh_replacement_and_logout() -> Result<()> {
     skip_if_no_network!(Ok(()));
 
     let _client_id_guard = EnvGuard::set(CLIENT_ID_OVERRIDE_ENV_VAR, "staging-client".to_string());
@@ -400,6 +400,82 @@ async fn refresh_token_succeeds_updates_storage() -> Result<()> {
         before.login_generation + 2
     );
     assert!(!grant.check_current().await?);
+    server.verify().await;
+    server.reset().await;
+    for login_owner_id in [Some("login-during-refresh"), None] {
+        let grant = ctx
+            .auth_manager
+            .lease_account_oauth_grant()
+            .await?
+            .context("replacement login grant")?
+            .release();
+        let mut latest_login = replacement.clone();
+        let tokens = latest_login
+            .tokens
+            .as_mut()
+            .context("latest login tokens")?;
+        tokens.login_owner_id = login_owner_id.map(str::to_owned);
+        if login_owner_id.is_none() {
+            tokens.oauth_provenance = None;
+        }
+        let saved_login = latest_login.clone();
+        let home = ctx.codex_home.path().to_path_buf();
+        Mock::given(method("POST"))
+            .and(path("/oauth/token"))
+            .respond_with(move |_: &wiremock::Request| {
+                // Replace the login or drop legacy metadata while refresh is in flight.
+                save_auth(
+                    &home,
+                    &saved_login,
+                    AuthCredentialsStoreMode::File,
+                    AuthKeyringBackendKind::default(),
+                )
+                .expect("replacement login saves");
+                ResponseTemplate::new(200).set_body_json(json!({
+                    "access_token": "stale-access-token", "refresh_token": "stale-refresh-token"
+                }))
+            })
+            .expect(1)
+            .mount(&server)
+            .await;
+        let refreshed = ctx.auth_manager.refresh_token_from_authority().await;
+        if login_owner_id.is_some() {
+            assert!(refreshed.is_err());
+        } else {
+            refreshed?;
+            let tokens = latest_login
+                .tokens
+                .as_mut()
+                .context("rotated legacy tokens")?;
+            tokens.access_token = "stale-access-token".into();
+            tokens.refresh_token = "stale-refresh-token".into();
+            tokens.oauth_provenance = initial_tokens.oauth_provenance.clone();
+            assert!(
+                ctx.auth_manager
+                    .lease_account_oauth_grant()
+                    .await?
+                    .is_none()
+            );
+        }
+        let stored = ctx.load_auth()?;
+        if login_owner_id.is_none() {
+            assert!(stored.last_refresh >= latest_login.last_refresh);
+            latest_login.last_refresh = stored.last_refresh;
+        }
+        assert_eq!(stored, latest_login);
+        assert!(!grant.check_current().await?);
+        ctx.auth_manager.reload().await;
+        assert_eq!(
+            ctx.auth_manager
+                .auth_cached()
+                .context("new login cached")?
+                .get_token_data()?,
+            latest_login.tokens.context("new login tokens")?
+        );
+        server.verify().await;
+        server.reset().await;
+    }
+    ctx.write_auth(&replacement).await?;
     let lease = ctx
         .auth_manager
         .lease_account_oauth_grant()
@@ -1441,21 +1517,15 @@ async fn refresh_token_reloads_changed_auth_after_permanent_failure() -> Result<
 
 #[serial_test::serial(auth_env)]
 #[tokio::test]
-async fn refresh_token_preserves_credentials_on_server_or_decode_failure() -> Result<()> {
+async fn refresh_token_preserves_credentials_on_refresh_failure() -> Result<()> {
     skip_if_no_network!(Ok(()));
 
-    for (status, body) in [
-        (500, json!({"error": "temporary-failure"})),
-        (200, json!(INITIAL_REFRESH_TOKEN)),
+    for (status, body, storage_failure) in [
+        (500, json!({"error": "temporary-failure"}), false),
+        (200, json!(INITIAL_REFRESH_TOKEN), false),
+        (200, json!({"access_token": "refreshed-access-token"}), true),
     ] {
         let server = MockServer::start().await;
-        Mock::given(method("POST"))
-            .and(path("/oauth/token"))
-            .respond_with(ResponseTemplate::new(status).set_body_json(body))
-            .expect(1)
-            .mount(&server)
-            .await;
-
         let ctx = RefreshTokenTestContext::new(&server).await?;
         let initial_last_refresh = Utc::now() - Duration::days(1);
         let initial_tokens = build_tokens(INITIAL_ACCESS_TOKEN, INITIAL_REFRESH_TOKEN);
@@ -1470,6 +1540,21 @@ async fn refresh_token_preserves_credentials_on_server_or_decode_failure() -> Re
             bedrock_access_keys: None,
         };
         ctx.write_auth(&initial_auth).await?;
+        let auth_path = ctx.codex_home.path().join("auth.json");
+        let unavailable_auth = auth_path.clone();
+        Mock::given(method("POST"))
+            .and(path("/oauth/token"))
+            .respond_with(move |_: &wiremock::Request| {
+                if storage_failure {
+                    // Make persistence and readback fail after refresh is admitted.
+                    std::fs::remove_file(&unavailable_auth).expect("remove fixture auth");
+                    std::fs::create_dir(&unavailable_auth).expect("make auth unavailable");
+                }
+                ResponseTemplate::new(status).set_body_json(body.clone())
+            })
+            .expect(1)
+            .mount(&server)
+            .await;
 
         let err = ctx
             .auth_manager
@@ -1481,8 +1566,9 @@ async fn refresh_token_preserves_credentials_on_server_or_decode_failure() -> Re
         assert_eq!(err.failed_reason(), None);
         assert!(!format!("{err} {err:?}").contains(INITIAL_REFRESH_TOKEN));
 
-        let stored = ctx.load_auth()?;
-        assert_eq!(stored, initial_auth);
+        if !storage_failure {
+            assert_eq!(ctx.load_auth()?, initial_auth);
+        }
         let cached_auth = ctx
             .auth_manager
             .auth()
@@ -1492,6 +1578,11 @@ async fn refresh_token_preserves_credentials_on_server_or_decode_failure() -> Re
             .get_token_data()
             .context("token data should remain cached")?;
         assert_eq!(cached, initial_tokens);
+        if storage_failure {
+            std::fs::remove_dir(&auth_path)?;
+            ctx.write_auth(&initial_auth).await?;
+            assert_eq!(ctx.load_auth()?, initial_auth);
+        }
 
         server.verify().await;
     }