fix(openai): persist authoritative quota exhaustion

Co-authored-by: Codex <noreply@openai.com>

backend/crates/providers/openai/src/credential/quota/mod.rs

@@ -36,7 +36,10 @@ use crate::transport::{CodexBackendClient, CodexClientError, CodexRequestContext
 
 use super::agent_identity::{CodexAgentIdentityTaskService, PreparedCodexRuntimeCredential};
 use super::repository::{CodexCredentialRepository, CredentialRepositoryError};
-use recovery::{quota_state_reason, quota_state_transition, quota_success_state};
+use recovery::{
+    authoritative_quota_snapshot_state, quota_state_reason, quota_state_transition,
+    quota_success_state,
+};
 use snapshot::{
     parse_account_quota_snapshot, quota_projection_ttl, quota_snapshot_from_observation,
     scheduling_signals_from_snapshot,
@@ -693,7 +696,7 @@ impl CodexCredentialQuotaService {
         } else {
             summary.updated += 1;
         }
-        if let Some(availability) = quota_success_state(
+        if let Some(availability) = authoritative_quota_snapshot_state(
             current.availability(),
             &refreshed_snapshot,
             previous_reset_at,
@@ -1021,7 +1024,7 @@ impl CodexCredentialQuotaService {
             return Err(CodexCredentialQuotaError::RevisionConflict);
         }
         self.scheduling.observe(&snapshot);
-        if let Some(availability) = quota_success_state(
+        if let Some(availability) = authoritative_quota_snapshot_state(
             current.availability(),
             &refreshed_snapshot,
             previous_reset_at,

backend/crates/providers/openai/src/credential/quota/recovery.rs

@@ -1,6 +1,7 @@
-//! 确认额度耗尽后的恢复证据与 availability 转换。
+//! 权威额度快照的 availability 转换。
 //!
-//! `QuotaExhausted -> Ready` 只能由同一 credential revision 的权威 usage 快照证明：
+//! 已触顶的 core primary 会立即把可调度账号标为 `QuotaExhausted`；
+//! `QuotaExhausted -> Ready` 则只能由同一 credential revision 的权威 usage 快照证明：
 //! 新快照未耗尽、core primary reset 已推进到下一窗口，且该窗口用量低于 10%。
 
 use std::time::SystemTime;
@@ -14,6 +15,31 @@ use crate::transport::CodexClientError;
 
 const RESET_RECOVERY_MAX_USED_PERCENT: f64 = 10.0;
 
+/// 基于成功的权威 usage 快照收敛可调度账号的额度状态。
+///
+/// 只接受上游明确拒绝使用且 core primary 触顶的证据；secondary、附加桶、
+/// credits 与 spend control 都不能单独把整个账号标成额度耗尽。
+pub(crate) fn authoritative_quota_snapshot_state(
+    current: AccountAvailability,
+    snapshot: &CodexAccountQuotaSnapshot,
+    previous_reset_at: Option<SystemTime>,
+) -> Option<AccountAvailability> {
+    match current {
+        AccountAvailability::Ready | AccountAvailability::Unknown
+            if snapshot.authoritative_core_primary_exhausted() =>
+        {
+            Some(AccountAvailability::QuotaExhausted)
+        }
+        AccountAvailability::QuotaExhausted => {
+            quota_success_state(current, snapshot, previous_reset_at)
+        }
+        AccountAvailability::Ready | AccountAvailability::Unknown => None,
+        AccountAvailability::Expired
+        | AccountAvailability::Banned
+        | AccountAvailability::Invalid => None,
+    }
+}
+
 /// 只处理 `QuotaExhausted` 的恢复。`allowed=true`、成功响应和百分比回落都不能单独覆盖
 /// 一次已确认的额度拒绝；它们在 reset 未推进时可能只是上游滞后的观测。
 ///

backend/crates/providers/openai/src/credential/quota/snapshot.rs

@@ -105,6 +105,7 @@ pub struct CodexAccountQuotaSnapshot {
     credential_revision: CredentialRevision,
     observed_at: SystemTime,
     fact: CodexQuotaFact,
+    authoritative_core_primary_exhausted: bool,
     windows: Vec<CodexQuotaWindow>,
 }
 
@@ -141,6 +142,12 @@ impl CodexAccountQuotaSnapshot {
             .find(|window| window.key() == CORE_PRIMARY_WINDOW_KEY)
     }
 
+    /// 上游成功 usage 明确拒绝请求，且 canonical core primary 已确认触顶。
+    #[must_use]
+    pub(crate) const fn authoritative_core_primary_exhausted(&self) -> bool {
+        self.authoritative_core_primary_exhausted
+    }
+
     /// 投影滚动已过期窗口：`reset_at` 已过的窗口 `used_percent=0`、
     /// `limit_reached=false`、`reset_at` 滚到下一个周期（reset 到期自动归零）。
     /// 只作用于展示/调度投影，不写回持久化 raw JSON。
@@ -316,6 +323,7 @@ pub(crate) fn parse_account_quota_snapshot(
     let object = usage
         .as_object()
         .ok_or(CodexCredentialQuotaError::InvalidCredentialData)?;
+    let authoritative_core_primary_exhausted = authoritative_core_primary_exhausted(object)?;
     let mut windows = Vec::new();
     for limit in canonical_rate_limits(object)? {
         parse_rate_limit_windows(
@@ -332,10 +340,55 @@ pub(crate) fn parse_account_quota_snapshot(
         credential_revision,
         observed_at,
         fact,
+        authoritative_core_primary_exhausted,
         windows,
     })
 }
 
+/// 账号状态只能由顶层 canonical Codex 额度桶的明确拒绝来改变。
+///
+/// `fact.exhausted()` 和窗口 `limit_reached` 都会聚合或继承 secondary、附加桶等事实，
+/// 因此不能单独用来将整个账号标为 `QuotaExhausted`。
+fn authoritative_core_primary_exhausted(
+    usage: &Map<String, Value>,
+) -> Result<bool, CodexCredentialQuotaError> {
+    let Some(rate_limit) = usage.get("rate_limit") else {
+        return Ok(false);
+    };
+    if rate_limit.is_null() {
+        return Ok(false);
+    }
+    let rate_limit = rate_limit
+        .as_object()
+        .ok_or(CodexCredentialQuotaError::InvalidCredentialData)?;
+    if optional_bool(rate_limit, "allowed")? != Some(false) {
+        return Ok(false);
+    }
+    let Some(primary_window) = rate_limit.get("primary_window") else {
+        return Ok(false);
+    };
+    if primary_window.is_null() {
+        return Ok(false);
+    }
+    let primary_window = primary_window
+        .as_object()
+        .ok_or(CodexCredentialQuotaError::InvalidCredentialData)?;
+    let primary_window_reached = optional_bool(rate_limit, "limit_reached")?.unwrap_or(false)
+        || optional_bool(primary_window, "limit_reached")?.unwrap_or(false)
+        || primary_window
+            .get("used_percent")
+            .map(|value| {
+                value
+                    .as_f64()
+                    .filter(|value| value.is_finite() && *value >= 0.0)
+                    .map(|value| value >= 100.0)
+                    .ok_or(CodexCredentialQuotaError::InvalidCredentialData)
+            })
+            .transpose()?
+            .unwrap_or(false);
+    Ok(primary_window_reached)
+}
+
 /// 一个规范化后的限流桶：`key` 用于稳定区分窗口，`source` 用于展示分组。
 struct CanonicalRateLimit<'a> {
     key: String,

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

@@ -487,7 +487,7 @@ async fn passive_rate_limit_headers_replace_stale_secondary_window() {
 }
 
 #[tokio::test]
-async fn manual_quota_refresh_drops_empty_secondary_window() {
+async fn manual_quota_refresh_marks_ready_account_exhausted_and_drops_empty_secondary_window() {
     let store = Arc::new(MemoryAccountStore::default());
     let account_id = "acct_manual_secondary_placeholder";
     create_account(&store, account_id).await;
@@ -533,6 +533,13 @@ async fn manual_quota_refresh_drops_empty_secondary_window() {
 
     assert_eq!(snapshot.windows().len(), 1);
     assert_eq!(snapshot.windows()[0].used_percent(), Some(100.0));
+    assert_eq!(
+        store
+            .account(account_id)
+            .expect("account after refresh")
+            .availability(),
+        AccountAvailability::QuotaExhausted
+    );
 }
 
 #[tokio::test]
@@ -824,7 +831,7 @@ async fn periodic_quota_synchronization_does_not_scan_stale_ready_accounts() {
 }
 
 #[tokio::test]
-async fn periodic_quota_synchronization_preserves_availability_when_usage_peaks() {
+async fn periodic_quota_synchronization_marks_ready_account_exhausted_when_primary_is_exhausted() {
     let store = Arc::new(MemoryAccountStore::default());
     create_account(&store, "acct_periodic_stuck_cooldown").await;
     let account = store
@@ -871,6 +878,50 @@ async fn periodic_quota_synchronization_preserves_availability_when_usage_peaks(
                 "primary_window": {"used_percent": 100, "reset_at": reset_at}
             }
         })))
+        .expect(1)
+        .mount(&server)
+        .await;
+    let service = quota_service_with_base_url(
+        &store,
+        reqwest::Client::builder()
+            .no_proxy()
+            .build()
+            .expect("client"),
+        server.uri(),
+    );
+    service
+        .prepare_scheduling(std::slice::from_ref(&account))
+        .await;
+
+    let summary = service.synchronize().await.expect("quota synchronization");
+
+    assert_eq!(summary.exhausted, 1);
+    let reconciled = store
+        .account("acct_periodic_stuck_cooldown")
+        .expect("reconciled account");
+    assert_eq!(
+        reconciled.availability(),
+        AccountAvailability::QuotaExhausted
+    );
+}
+
+#[tokio::test]
+async fn periodic_quota_synchronization_keeps_ready_when_only_secondary_window_is_exhausted() {
+    let store = Arc::new(MemoryAccountStore::default());
+    create_account(&store, "acct_periodic_secondary_only").await;
+    let reset_at = Utc::now().timestamp() + 24 * 60 * 60;
+    let server = MockServer::start().await;
+    Mock::given(method("GET"))
+        .and(path("/api/codex/usage"))
+        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
+            "rate_limit": {
+                "allowed": true,
+                "limit_reached": false,
+                "primary_window": {"used_percent": 5, "reset_at": reset_at},
+                "secondary_window": {"used_percent": 100, "reset_at": reset_at}
+            }
+        })))
+        .expect(1)
         .mount(&server)
         .await;
     let service = quota_service_with_base_url(
@@ -882,13 +933,99 @@ async fn periodic_quota_synchronization_preserves_availability_when_usage_peaks(
         server.uri(),
     );
 
-    service.synchronize().await.expect("quota synchronization");
+    let summary = service.synchronize().await.expect("quota synchronization");
 
-    // 429/触顶不再写入 availability（限流不改变账号可用性），worker 只更新 quota 数据。
-    let reconciled = store
-        .account("acct_periodic_stuck_cooldown")
-        .expect("reconciled account");
-    assert_eq!(reconciled.availability(), AccountAvailability::Ready);
+    assert_eq!(summary.exhausted, 1);
+    assert_eq!(
+        store
+            .account("acct_periodic_secondary_only")
+            .expect("reconciled account")
+            .availability(),
+        AccountAvailability::Ready
+    );
+}
+
+#[tokio::test]
+async fn periodic_quota_synchronization_keeps_ready_when_primary_is_full_but_requests_are_allowed()
+{
+    let store = Arc::new(MemoryAccountStore::default());
+    create_account(&store, "acct_periodic_primary_full_but_allowed").await;
+    let reset_at = Utc::now().timestamp() + 24 * 60 * 60;
+    let server = MockServer::start().await;
+    Mock::given(method("GET"))
+        .and(path("/api/codex/usage"))
+        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
+            "rate_limit": {
+                "allowed": true,
+                "limit_reached": false,
+                "primary_window": {"used_percent": 100, "reset_at": reset_at}
+            }
+        })))
+        .expect(1)
+        .mount(&server)
+        .await;
+    let service = quota_service_with_base_url(
+        &store,
+        reqwest::Client::builder()
+            .no_proxy()
+            .build()
+            .expect("client"),
+        server.uri(),
+    );
+
+    let summary = service.synchronize().await.expect("quota synchronization");
+
+    assert_eq!(summary.exhausted, 1);
+    assert_eq!(
+        store
+            .account("acct_periodic_primary_full_but_allowed")
+            .expect("reconciled account")
+            .availability(),
+        AccountAvailability::Ready
+    );
+}
+
+#[tokio::test]
+async fn periodic_quota_synchronization_keeps_ready_without_primary_exhaustion_evidence() {
+    let store = Arc::new(MemoryAccountStore::default());
+    create_account(
+        &store,
+        "acct_periodic_disallowed_without_primary_exhaustion",
+    )
+    .await;
+    let reset_at = Utc::now().timestamp() + 24 * 60 * 60;
+    let server = MockServer::start().await;
+    Mock::given(method("GET"))
+        .and(path("/api/codex/usage"))
+        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
+            "rate_limit": {
+                "allowed": false,
+                "limit_reached": false,
+                "primary_window": {"used_percent": 5, "reset_at": reset_at}
+            }
+        })))
+        .expect(1)
+        .mount(&server)
+        .await;
+    let service = quota_service_with_base_url(
+        &store,
+        reqwest::Client::builder()
+            .no_proxy()
+            .build()
+            .expect("client"),
+        server.uri(),
+    );
+
+    let summary = service.synchronize().await.expect("quota synchronization");
+
+    assert_eq!(summary.exhausted, 1);
+    assert_eq!(
+        store
+            .account("acct_periodic_disallowed_without_primary_exhaustion")
+            .expect("reconciled account")
+            .availability(),
+        AccountAvailability::Ready
+    );
 }
 
 #[tokio::test]