fix(openai): require low usage for quota recovery

Co-authored-by: Codex <noreply@openai.com>
2026-11-06 08:06:19 +00:00 · 2026-08-10 15:55:58 +08:00
parent 44a2a87b69
commit 5b7b9b21ee
5 changed files with 172 additions and 51 deletions
@@ -397,7 +397,7 @@ impl CodexCredentialQuotaService {
        }
    }

-    /// 写入新快照前读取已确认耗尽窗口的 reset，用于判断窗口是否真的推进。
+    /// 写入新快照前读取 core primary 已确认耗尽窗口的 reset，用于判断该窗口是否真的推进。
    async fn previous_confirmed_reset(&self, account: &ProviderAccount) -> Option<SystemTime> {
        let observations = self
            .store
@@ -409,7 +409,10 @@ impl CodexCredentialQuotaService {
                && observation.expected_revision == account.revision()
        })?;
        let snapshot = quota_snapshot_from_observation(&observation)?;
-        snapshot.fact().resets_at().map(SystemTime::from)
+        snapshot
+            .core_primary_window()
+            .and_then(|window| window.reset_at())
+            .map(SystemTime::from)
    }

    /// 将已确认的额度拒绝投影为 100%，使展示、调度和恢复共用同一窗口事实。
@@ -633,7 +636,6 @@ impl CodexCredentialQuotaService {
            observed_at,
            value,
        )?;
-        let refreshed_fact = refreshed_snapshot.fact();
        let previous_reset_at = self.previous_confirmed_reset(account).await;
        let object = normalize_quota_window_placeholders(
            value
@@ -642,12 +644,16 @@ impl CodexCredentialQuotaService {
                .ok_or(CodexCredentialQuotaError::InvalidCredentialData)?,
        );
        let object = if account.availability() == AccountAvailability::QuotaExhausted
-            && quota_success_state(account.availability(), refreshed_fact, previous_reset_at)
-                .is_none()
+            && quota_success_state(
+                account.availability(),
+                &refreshed_snapshot,
+                previous_reset_at,
+            )
+            .is_none()
        {
            confirmed_exhaustion_projection(
                object,
-                confirmed_exhaustion_reset_at(previous_reset_at, refreshed_fact),
+                confirmed_exhaustion_reset_at(previous_reset_at, &refreshed_snapshot),
            )
        } else {
            object
@@ -687,9 +693,11 @@ impl CodexCredentialQuotaService {
        } else {
            summary.updated += 1;
        }
-        if let Some(availability) =
-            quota_success_state(current.availability(), refreshed_fact, previous_reset_at)
-        {
+        if let Some(availability) = quota_success_state(
+            current.availability(),
+            &refreshed_snapshot,
+            previous_reset_at,
+        ) {
            let _ = self
                .repository
                .apply_state(&current, availability, observed_at)
@@ -755,13 +763,18 @@ impl CodexCredentialQuotaService {
        }
        let quota = merge_passive_quota(existing, &rate_limits);
        let observed_at = SystemTime::now();
-        let refreshed_fact = parse_codex_quota_usage(&Value::Object(quota.clone()))?;
+        let refreshed_snapshot = parse_account_quota_snapshot(
+            account.id().clone(),
+            account.revision(),
+            observed_at,
+            &Value::Object(quota.clone()),
+        )?;
        let quota = if account.availability() == AccountAvailability::QuotaExhausted {
            // 被动成功响应不能解除确认额度耗尽；继续展示 100%，直到权威 usage 快照
            // 证明旧 reset 已跨过且新窗口已经推进。
            confirmed_exhaustion_projection(
                quota,
-                confirmed_exhaustion_reset_at(previous_reset_at, refreshed_fact),
+                confirmed_exhaustion_reset_at(previous_reset_at, &refreshed_snapshot),
            )
        } else {
            quota
@@ -957,7 +970,6 @@ impl CodexCredentialQuotaService {
            observed_at,
            &value,
        )?;
-        let refreshed_fact = refreshed_snapshot.fact();
        let previous_reset_at = self.previous_confirmed_reset(&account).await;
        let object = normalize_quota_window_placeholders(
            value
@@ -966,12 +978,16 @@ impl CodexCredentialQuotaService {
                .ok_or(CodexCredentialQuotaError::InvalidCredentialData)?,
        );
        let object = if account.availability() == AccountAvailability::QuotaExhausted
-            && quota_success_state(account.availability(), refreshed_fact, previous_reset_at)
-                .is_none()
+            && quota_success_state(
+                account.availability(),
+                &refreshed_snapshot,
+                previous_reset_at,
+            )
+            .is_none()
        {
            confirmed_exhaustion_projection(
                object,
-                confirmed_exhaustion_reset_at(previous_reset_at, refreshed_fact),
+                confirmed_exhaustion_reset_at(previous_reset_at, &refreshed_snapshot),
            )
        } else {
            object
@@ -1005,9 +1021,11 @@ impl CodexCredentialQuotaService {
            return Err(CodexCredentialQuotaError::RevisionConflict);
        }
        self.scheduling.observe(&snapshot);
-        if let Some(availability) =
-            quota_success_state(current.availability(), refreshed_fact, previous_reset_at)
-        {
+        if let Some(availability) = quota_success_state(
+            current.availability(),
+            &refreshed_snapshot,
+            previous_reset_at,
+        ) {
            self.repository
                .apply_state(&current, availability, observed_at)
                .await?;
@@ -1314,11 +1332,16 @@ fn confirmed_exhaustion_projection(
 /// 否则保留已确认失败的旧窗口，避免 98/99% 快照反向解锁。
 fn confirmed_exhaustion_reset_at(
    previous_reset_at: Option<SystemTime>,
-    refreshed_fact: CodexQuotaFact,
+    refreshed_snapshot: &CodexAccountQuotaSnapshot,
 ) -> Option<i64> {
-    let refreshed_reset_at = refreshed_fact.resets_at().map(SystemTime::from);
+    let refreshed_reset_at = refreshed_snapshot
+        .core_primary_window()
+        .and_then(|window| window.reset_at())
+        .map(SystemTime::from);
    let reset_at = match (previous_reset_at, refreshed_reset_at) {
-        (Some(previous), Some(refreshed)) if refreshed_fact.exhausted() && refreshed > previous => {
+        (Some(previous), Some(refreshed))
+            if refreshed_snapshot.fact().exhausted() && refreshed > previous =>
+        {
            refreshed
        }
        (Some(previous), _) => previous,
@@ -1,7 +1,7 @@
 //! 确认额度耗尽后的恢复证据与 availability 转换。
 //!
 //! `QuotaExhausted -> Ready` 只能由同一 credential revision 的权威 usage 快照证明：
-//! 新快照未耗尽，且 reset 已推进到下一窗口。
+//! 新快照未耗尽、core primary reset 已推进到下一窗口，且该窗口用量低于 10%。

 use std::time::SystemTime;

@@ -9,25 +9,30 @@ use gateway_core::engine::credential::AccountAvailability;
 use reqwest::StatusCode;
 use serde_json::Value;

-use super::snapshot::CodexQuotaFact;
+use super::snapshot::CodexAccountQuotaSnapshot;
 use crate::transport::CodexClientError;

+const RESET_RECOVERY_MAX_USED_PERCENT: f64 = 10.0;
+
 /// 只处理 `QuotaExhausted` 的恢复。`allowed=true`、成功响应和百分比回落都不能单独覆盖
 /// 一次已确认的额度拒绝；它们在 reset 未推进时可能只是上游滞后的观测。
 ///
-/// `previous_reset_at` 是写入新快照前读取的已确认耗尽窗口。恢复只要求权威快照的 reset
-/// 指向比旧窗口更晚的新周期，不等待旧 reset 的墙钟时间到达。
+/// `previous_reset_at` 是写入新快照前读取的 core primary 已确认耗尽窗口。上游可能只
+/// 调整 reset 时间而尚未重置额度，因此还要求同一新鲜 primary 窗口的用量低于 10%。
 pub(crate) fn quota_success_state(
    current: AccountAvailability,
-    fact: CodexQuotaFact,
+    snapshot: &CodexAccountQuotaSnapshot,
    previous_reset_at: Option<SystemTime>,
 ) -> Option<AccountAvailability> {
-    if current != AccountAvailability::QuotaExhausted || fact.exhausted() {
+    if current != AccountAvailability::QuotaExhausted || snapshot.fact().exhausted() {
        return None;
    }
    let previous_reset_at = previous_reset_at?;
-    let refreshed_reset_at = fact.resets_at().map(SystemTime::from)?;
-    (refreshed_reset_at > previous_reset_at).then_some(AccountAvailability::Ready)
+    let primary_window = snapshot.core_primary_window()?;
+    let refreshed_reset_at = primary_window.reset_at().map(SystemTime::from)?;
+    let used_percent = primary_window.used_percent()?;
+    (refreshed_reset_at > previous_reset_at && used_percent < RESET_RECOVERY_MAX_USED_PERCENT)
+        .then_some(AccountAvailability::Ready)
 }

 pub(crate) fn quota_state_transition(error: &CodexClientError) -> Option<AccountAvailability> {
@@ -13,6 +13,8 @@ use serde_json::{Map, Value};

 use super::{CodexCredentialQuotaError, QUOTA_SCHEDULING_TTL};

+const CORE_PRIMARY_WINDOW_KEY: &str = "core-primary";
+
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub struct CodexQuotaFact {
    remaining_percent: Option<u8>,
@@ -132,6 +134,13 @@ impl CodexAccountQuotaSnapshot {
        &self.windows
    }

+    #[must_use]
+    pub(crate) fn core_primary_window(&self) -> Option<&CodexQuotaWindow> {
+        self.windows
+            .iter()
+            .find(|window| window.key() == CORE_PRIMARY_WINDOW_KEY)
+    }
+
    /// 投影滚动已过期窗口：`reset_at` 已过的窗口 `used_percent=0`、
    /// `limit_reached=false`、`reset_at` 滚到下一个周期（reset 到期自动归零）。
    /// 只作用于展示/调度投影，不写回持久化 raw JSON。
@@ -1254,7 +1254,7 @@ async fn periodic_quota_synchronization_rechecks_exhausted_accounts_without_wait
 }

 #[tokio::test]
-async fn periodic_quota_synchronization_does_not_recover_from_percent_drop_before_reset() {
+async fn periodic_quota_synchronization_keeps_exhaustion_at_ten_percent_after_advanced_reset() {
    let store = Arc::new(MemoryAccountStore::default());
    let account_id = "acct_periodic_recovery_evidence";
    create_account(&store, account_id).await;
@@ -1333,8 +1333,8 @@ async fn periodic_quota_synchronization_does_not_recover_from_percent_drop_befor
        .and(path("/api/codex/usage"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
            "rate_limit": {
-                // 无 allowed/limit_reached 标记的模糊快照。
-                "primary_window": {"used_percent": 99, "reset_at": reset_at}
+                // reset 已后移，但 10% 仍不能证明这是刚重置的新窗口。
+                "primary_window": {"used_percent": 10, "reset_at": reset_at + 3600}
            }
        })))
        .expect(1)
@@ -1350,19 +1350,19 @@ async fn periodic_quota_synchronization_does_not_recover_from_percent_drop_befor
        server.uri(),
    );

-    let lower_percent = next_cycle
+    let ten_percent = next_cycle
        .synchronize()
        .await
-        .expect("lower percent quota recheck");
+        .expect("ten-percent quota recheck");

-    assert_eq!(lower_percent.exhausted, 1);
+    assert_eq!(ten_percent.exhausted, 1);
    assert_eq!(
        store
            .account(account_id)
-            .expect("account after lower percent snapshot")
+            .expect("account after ten-percent snapshot")
            .availability(),
        AccountAvailability::QuotaExhausted,
-        "a provider percentage change is not sufficient to release a confirmed exhaustion"
+        "an advanced reset at exactly 10% is not sufficient to release confirmed exhaustion"
    );
    let snapshot = next_cycle
        .read_account(account.id())
@@ -1373,12 +1373,12 @@ async fn periodic_quota_synchronization_does_not_recover_from_percent_drop_befor
 }

 #[tokio::test]
-async fn periodic_quota_synchronization_recovers_after_recorded_reset() {
+async fn periodic_quota_synchronization_recovers_with_advanced_reset_and_low_primary_usage() {
    let store = Arc::new(MemoryAccountStore::default());
    let account_id = "acct_periodic_reset_elapsed";
    create_account(&store, account_id).await;
    let account = store.account(account_id).expect("created account");
-    let elapsed_reset = Utc::now().timestamp() - 60;
+    let old_reset = Utc::now().timestamp() + 3_600;
    store
        .compare_and_swap_quota(QuotaObservation {
            account_id: account.id().clone(),
@@ -1388,7 +1388,7 @@ async fn periodic_quota_synchronization_recovers_after_recorded_reset() {
                    "rate_limit": {
                        "allowed": false,
                        "limit_reached": true,
-                        "primary_window": {"used_percent": 100, "reset_at": elapsed_reset}
+                        "primary_window": {"used_percent": 100, "reset_at": old_reset}
                    }
                })
                .as_object()
@@ -1399,7 +1399,7 @@ async fn periodic_quota_synchronization_recovers_after_recorded_reset() {
            limit_reached: Some(true),
        })
        .await
-        .expect("record elapsed exhausted reset");
+        .expect("record future exhausted reset");
    store
        .apply_state_change(AccountStateChange {
            message: None,
@@ -1409,9 +1409,9 @@ async fn periodic_quota_synchronization_recovers_after_recorded_reset() {
            observed_at: SystemTime::now(),
        })
        .await
-        .expect("mark account exhausted with an elapsed reset");
+        .expect("mark account exhausted with a future reset");
    let server = MockServer::start().await;
-    let next_reset = Utc::now().timestamp() + 3600;
+    let next_reset = old_reset + 3_600;
    Mock::given(method("GET"))
        .and(path("/api/codex/usage"))
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
@@ -1433,7 +1433,7 @@ async fn periodic_quota_synchronization_recovers_after_recorded_reset() {
        server.uri(),
    );

-    // 定时复核可提前执行，但只有已记录的 reset 到期后才释放额度耗尽状态。
+    // 定时复核可在旧 reset 前执行；新主窗口已推进且用量低于 10% 才释放额度耗尽状态。
    let summary = service.synchronize().await.expect("quota synchronization");

    assert_eq!(summary.updated, 1);
@@ -1585,7 +1585,7 @@ async fn invalid_quota_json_for_one_account_does_not_abort_the_batch() {
        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
            "rate_limit": {
                "allowed": true,
-                "primary_window": {"used_percent": 25.0, "reset_at": next_reset}
+                "primary_window": {"used_percent": 1.0, "reset_at": next_reset}
            }
        })))
        .mount(&server)
@@ -1614,5 +1614,5 @@ async fn invalid_quota_json_for_one_account_does_not_abort_the_batch() {
        .await
        .expect("read good quota")
        .expect("good quota persisted");
-    assert_eq!(snapshot.fact().remaining_percent(), Some(75));
+    assert_eq!(snapshot.fact().remaining_percent(), Some(99));
 }
@@ -1,6 +1,6 @@
 //! 402 恢复回归。
 //!
-//! 覆盖 `QuotaExhausted -> Ready` 的证据规则：新 reset 后移 + 新快照未耗尽。
+//! 覆盖 `QuotaExhausted -> Ready` 的证据规则：新 core primary reset 后移 + 新快照用量低于 10%。

 use super::*;

@@ -127,8 +127,8 @@ async fn manual_quota_refresh_keeps_confirmed_exhaustion_until_the_recorded_rese
 }

 #[tokio::test]
-async fn advanced_reset_recovers_ready_before_the_recorded_reset_time() {
-    // 旧 reset 仍在未来；只要新窗口 reset 后移且额度正常，仍必须恢复 Ready。
+async fn advanced_reset_with_primary_usage_below_ten_percent_recovers_ready() {
+    // 旧 reset 仍在未来；同一主窗口 reset 后移且用量低于 10% 时才恢复 Ready。
    let store = Arc::new(MemoryAccountStore::default());
    let account_id = "acct_cross_window_recovery";
    create_account(&store, account_id).await;
@@ -181,7 +181,7 @@ async fn advanced_reset_recovers_ready_before_the_recorded_reset_time() {
            "rate_limit": {
                "allowed": true,
                "limit_reached": false,
-                "primary_window": {"used_percent": 5, "reset_at": future_reset}
+                "primary_window": {"used_percent": 9.9, "reset_at": future_reset}
            }
        })))
        .mount(&server)
@@ -207,7 +207,91 @@ async fn advanced_reset_recovers_ready_before_the_recorded_reset_time() {
            .expect("account after cross-window recovery")
            .availability(),
        AccountAvailability::Ready,
-        "an advanced reset must recover QuotaExhausted before the recorded reset time"
+        "an advanced reset with a nearly unused primary window must recover QuotaExhausted"
+    );
+}
+
+#[tokio::test]
+async fn advanced_reset_with_primary_usage_at_ten_percent_keeps_confirmed_exhaustion() {
+    let store = Arc::new(MemoryAccountStore::default());
+    let account_id = "acct_cross_window_ten_percent";
+    create_account(&store, account_id).await;
+    let account = store.account(account_id).expect("created account");
+    let old_reset = Utc::now().timestamp() + 3_600;
+    store
+        .compare_and_swap_quota(QuotaObservation {
+            account_id: account.id().clone(),
+            expected_revision: account.revision(),
+            quota: Some(OpaqueProviderData::new(
+                json!({
+                    "rate_limit": {
+                        "allowed": false,
+                        "limit_reached": true,
+                        "primary_window": {
+                            "used_percent": 100,
+                            "reset_at": old_reset,
+                            "limit_window_seconds": 3600
+                        }
+                    }
+                })
+                .as_object()
+                .expect("quota object")
+                .clone(),
+            )),
+            observed_at: Some(SystemTime::now()),
+            limit_reached: None,
+        })
+        .await
+        .expect("seed old exhausted snapshot");
+    store
+        .apply_state_change(AccountStateChange {
+            message: None,
+            account_id: account.id().clone(),
+            expected_revision: account.revision(),
+            availability: AccountAvailability::QuotaExhausted,
+            observed_at: SystemTime::now(),
+        })
+        .await
+        .expect("mark account exhausted");
+    let server = MockServer::start().await;
+    let future_reset = old_reset + 3_600;
+    Mock::given(method("GET"))
+        .and(path("/api/codex/usage"))
+        .and(header(
+            "authorization",
+            format!("Bearer token-{account_id}"),
+        ))
+        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
+            "rate_limit": {
+                "allowed": true,
+                "limit_reached": false,
+                "primary_window": {"used_percent": 10.0, "reset_at": future_reset}
+            }
+        })))
+        .mount(&server)
+        .await;
+    let service = quota_service_with_base_url(
+        &store,
+        reqwest::Client::builder()
+            .no_proxy()
+            .build()
+            .expect("client"),
+        server.uri(),
+    );
+
+    let held = service
+        .refresh_account(account.id())
+        .await
+        .expect("refresh non-empty new window quota");
+
+    assert!(held.fact().exhausted());
+    assert_eq!(
+        store
+            .account(account_id)
+            .expect("account after ten-percent snapshot")
+            .availability(),
+        AccountAvailability::QuotaExhausted,
+        "a reset advance at exactly 10% must not release confirmed exhaustion"
    );
 }