fix(openai): keep account selection through quota observation updates (#400)

Successful responses keep refreshing the quota observation time of the
account they used. The post-selection credential reload compared the
whole account snapshot, so a busy account could hit three observation-only
conflicts in a row and fail with account_snapshot_conflict before any
upstream request was sent.

Accept the reloaded snapshot when only the observation time of a
non-exhausted quota moved. Credential, state, scheduling, scope and quota
conclusion changes still invalidate the selection and retry as before.

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

backend/crates/providers/openai/src/credential/repository.rs

@@ -180,6 +180,28 @@ impl CodexCredentialRepository {
         self.decode_runtime_credential(&loaded)
     }
 
+    /// 选号专用的凭据重读，返回最新账号快照
+    ///
+    /// 成功推理会持续刷新额度观测时刻；未耗尽账号的访问结论不变时调度资格不变，
+    /// 沿用最新快照，避免繁忙账号在选号阶段反复冲突
+    pub async fn reload_selected_credential(
+        &self,
+        account: &ProviderAccount,
+    ) -> Result<(ProviderAccount, CodexRuntimeCredential), CredentialRepositoryError> {
+        if account.provider().as_str() != PROVIDER_NAME {
+            return Err(CredentialRepositoryError::InvalidCredentialData(None));
+        }
+        let loaded = self
+            .store
+            .load_credential(account.id(), account.revision())
+            .await?;
+        if !only_quota_observation_advanced(account, &loaded.account) {
+            return Err(CredentialRepositoryError::RevisionConflict(None));
+        }
+        let runtime = self.decode_runtime_credential(&loaded)?;
+        Ok((loaded.account, runtime))
+    }
+
     pub async fn load_complete_data(
         &self,
         account: &ProviderAccount,
@@ -320,3 +342,27 @@ fn oauth_account_state(account: &ProviderAccount, observed: CredentialState) ->
         observed
     }
 }
+
+fn only_quota_observation_advanced(selected: &ProviderAccount, current: &ProviderAccount) -> bool {
+    if selected == current {
+        return true;
+    }
+    let (before, after) = (selected.quota(), current.quota());
+    // 耗尽状态的复核时间依赖观测时刻，必须按冲突处理
+    if before.is_exhausted()
+        || before.access() != after.access()
+        || before.evidence() != after.evidence()
+        || before.reset_at() != after.reset_at()
+    {
+        return false;
+    }
+    // 只替换额度观测，其余账号事实必须与选号时完全一致
+    let normalized = current.clone().with_account_facts(
+        current.enabled(),
+        current.credential_state(),
+        before,
+        current.last_error_reason(),
+        current.last_error_message().map(str::to_owned),
+    );
+    normalized == *selected
+}

backend/crates/providers/openai/src/credential/selector.rs

@@ -624,8 +624,8 @@ impl CodexCredentialSelector {
                 if request.requires_websocket
                     && pinned_account.as_ref().is_none_or(|id| id == account.id())
                 {
-                    let runtime = match self.repository.load_runtime_credential(&account).await {
-                        Ok(runtime) => runtime,
+                    let runtime = match self.repository.reload_selected_credential(&account).await {
+                        Ok((_, runtime)) => runtime,
                         Err(CredentialRepositoryError::RevisionConflict(_)) => {
                             retry_account_snapshot(
                                 request.attempt,
@@ -907,10 +907,15 @@ impl CodexCredentialSelector {
                     .find(|candidate| candidate.account.id() == selected.account.id())
                     .map(|candidate| candidate.account.clone())
                     .ok_or(CredentialSelectionError::InvalidCredential(None))?;
-                // 额度观测等并发更新会使整个账号快照失效，必须重新选号并校验资格
+                // 凭据、状态或额度结论的并发更新会使账号快照失效，必须重新选号并校验资格；
+                // 仅额度观测时刻前进时沿用最新快照
                 // 在占用租约和请求间隔前完成校验，避免重读被自己的异步释放挡住
-                let runtime = match self.repository.load_runtime_credential(&account).await {
-                    Ok(runtime) => runtime,
+                let (account, runtime) = match self
+                    .repository
+                    .reload_selected_credential(&account)
+                    .await
+                {
+                    Ok(reloaded) => reloaded,
                     Err(CredentialRepositoryError::RevisionConflict(_)) => {
                         retry_account_snapshot(request.attempt, &account, &mut snapshot_retries)?;
                         continue 'capacity;

backend/crates/providers/openai/tests/credential/contract/mod.rs

@@ -537,7 +537,11 @@ fn selector_reloads_quota_snapshot_before_acquiring_the_only_lease() {
         selected.account(),
         &store.account("acct_primary").expect("current account")
     );
-    assert_eq!(store.credential_loads(), 2);
+    assert_eq!(
+        store.credential_loads(),
+        1,
+        "observation-only quota updates must not restart selection"
+    );
     let requests = leases.requests.lock().expect("lease requests");
     assert_eq!(
         requests.len(),
@@ -548,6 +552,52 @@ fn selector_reloads_quota_snapshot_before_acquiring_the_only_lease() {
     assert_eq!(requests[0].request_interval(), Duration::from_millis(10));
 }
 
+#[test]
+fn selector_rejects_quota_exhaustion_observed_after_selection() {
+    let store = Arc::new(MemoryAccountStore::default());
+    create_account(&store, "acct_primary", "at-primary");
+    let original = store.account("acct_primary").expect("account");
+    block_on(store.apply_quota_access(QuotaAccessChange {
+        account_id: original.id().clone(),
+        expected_revision: original.revision(),
+        state: QuotaState::allowed(SystemTime::now()),
+    }))
+    .expect("initial quota observation");
+    store.on_credential_load(Arc::new(|store, id, count| {
+        Box::pin(async move {
+            if count == 1 {
+                let account = store.account(id.as_str()).expect("account");
+                store
+                    .apply_quota_access(QuotaAccessChange {
+                        account_id: id.clone(),
+                        expected_revision: account.revision(),
+                        state: QuotaState::exhausted(
+                            QuotaEvidence::UsageLimitReached,
+                            SystemTime::now(),
+                            None,
+                        ),
+                    })
+                    .await?;
+            }
+            Ok(())
+        })
+    }));
+    let leases = Arc::new(TestLeaseCoordinator::default());
+    let selector = selector(&store, Arc::clone(&leases));
+    let attempt = attempt(BTreeSet::new());
+    let error = block_on(selector.select(&SelectCodexCredential {
+        upstream_model: "gpt-5.4",
+        request_url: &Url::parse("https://chatgpt.com/backend-api/codex/responses").expect("URL"),
+        attempt: &attempt,
+        session_affinity_key: None,
+    }))
+    .expect_err("quota exhaustion must still invalidate the selected snapshot");
+
+    assert!(matches!(error, CredentialSelectionError::QuotaExhausted));
+    assert_eq!(store.credential_loads(), 1);
+    assert!(leases.requests.lock().expect("lease requests").is_empty());
+}
+
 #[test]
 fn selector_uses_rotated_credentials_after_a_snapshot_conflict() {
     use gateway_core::account::{CredentialCasOutcome, CredentialCasUpdate, ProviderAccountUpdate};

backend/crates/providers/openai/tests/provider/contract/mod.rs

@@ -2329,7 +2329,11 @@ async fn quota_snapshot_race_reloads_and_sends_the_request_once() {
         .await
         .expect("concurrent quota observation must not fail prepare");
     assert_eq!(stream.metadata().provider_account_id(), original.id());
-    assert_eq!(store.credential_loads(), 2);
+    assert_eq!(
+        store.credential_loads(),
+        1,
+        "observation-only quota updates must not restart selection"
+    );
     assert_eq!(leases.requests.lock().expect("leases").len(), 1);
     let mut completed = false;
     while let Some(event) = stream.next().await {
@@ -2476,19 +2480,7 @@ async fn repeated_snapshot_conflicts_are_bounded_and_report_the_selection_stage(
     store.on_credential_load(Arc::new(|store, id, count| {
         Box::pin(async move {
             assert!(count <= 4, "snapshot conflict retries must be bounded");
-            let account = store.account(id.as_str()).expect("account");
-            let observed_at = account
-                .quota()
-                .observed_at()
-                .unwrap_or_else(SystemTime::now)
-                + Duration::from_millis(1);
-            store
-                .apply_quota_access(QuotaAccessChange {
-                    account_id: id.clone(),
-                    expected_revision: account.revision(),
-                    state: QuotaState::allowed(observed_at),
-                })
-                .await?;
+            bump_account_weight(store, id, count);
             Ok(())
         })
     }));
@@ -2547,19 +2539,7 @@ async fn api_websocket_precheck_and_selection_share_the_snapshot_retry_budget()
             );
             // 第 1、4 次在传输预检冲突，第 3、6 次在最终候选校验冲突
             if matches!(count, 1 | 3 | 4 | 6) {
-                let account = store.account(id.as_str()).expect("account");
-                let observed_at = account
-                    .quota()
-                    .observed_at()
-                    .unwrap_or_else(SystemTime::now)
-                    + Duration::from_millis(1);
-                store
-                    .apply_quota_access(QuotaAccessChange {
-                        account_id: id.clone(),
-                        expected_revision: account.revision(),
-                        state: QuotaState::allowed(observed_at),
-                    })
-                    .await?;
+                bump_account_weight(store, id, count);
             }
             Ok(())
         })
@@ -2609,6 +2589,15 @@ async fn api_websocket_precheck_and_selection_share_the_snapshot_retry_budget()
     );
 }
 
+// 调度权重属于选号事实，每次变化都必须让已选快照失效
+fn bump_account_weight(store: &MemoryAccountStore, id: &ProviderAccountId, count: usize) {
+    store.set_scheduling(
+        id.as_str(),
+        None,
+        AccountWeight::new(u16::try_from(count + 1).expect("small count")).expect("weight"),
+    );
+}
+
 async fn exhaust_account_quota(store: &Arc<MemoryAccountStore>, account_id: &str) {
     let account = store.account(account_id).expect("test account");
     store