fix(openai): honor usable credits and recover quota after reset (#469)

Refs #465 Co-authored-by: Zyy <zyycn@users.noreply.github.com> Co-authored-by: Codex <noreply@openai.com>
2026-11-06 08:06:19 +00:00 · 2026-10-18 22:45:15 +08:00
parent 591f95cc18
commit cef9f0b21e
13 changed files with 735 additions and 59 deletions
@@ -73,6 +73,10 @@ const DEFAULT_RATE_LIMIT_COOLDOWN: Duration = Duration::from_secs(60);
 pub(crate) const QUOTA_SCHEDULING_TTL: Duration = Duration::from_secs(10 * 60);
 const QUOTA_HYDRATION_FAILURE_TTL: Duration = Duration::from_secs(5);
 const PERIODIC_QUOTA_REFRESH_RETRY_INTERVAL: Duration = Duration::from_secs(30 * 60);
+const QUOTA_RESET_FAST_REFRESH_WINDOW: Duration = Duration::from_secs(5 * 60);
+const QUOTA_RESET_FAST_REFRESH_INTERVAL: Duration = Duration::from_secs(30);
+const QUOTA_RESET_REFRESH_RETRY_INTERVAL: Duration = Duration::from_secs(5 * 60);
+const EXHAUSTED_QUOTA_RESET_GRACE: Duration = Duration::from_secs(10);
 const QUOTA_RESET_GRACE: Duration = Duration::from_secs(2 * 60);
 /// 首次 OAuth 异步观察失败时，由既有 quota worker 兜底重试的单轮上限
 /// 随机启动延迟已把候选摊到多轮，该上限仅作单轮安全边界
@@ -4,7 +4,7 @@ use std::collections::BTreeMap;
 use std::time::SystemTime;

 use chrono::{DateTime, Utc};
-use gateway_core::account::{QuotaEvidence, QuotaState};
+use gateway_core::account::{QuotaAccessState, QuotaEvidence, QuotaState};
 use serde::{Deserialize, Serialize};
 use serde_json::{Map, Value};

@@ -127,11 +127,8 @@ fn window_reset_recovered(window: &CodexQuotaWindow, previous: Option<DateTime<U
 }

 /// 统一手动刷新和 worker 的恢复规则
-/// 已耗尽窗口的解除有两条证据路径：
-/// reset 前进且用量低于低用量阈值（窗口已滚动），或同一窗口内连续两次观测
-/// 都未触顶（滑动窗口的用量回落，reset 不滚动）
-/// 后者防止账号被钉在旧
-/// reset 日期上；连续两次的要求排除耗尽后立刻拉到的旧快照
+/// 到期前仍要求窗口滚动且低用量，或同窗口连续两次未触顶
+/// 已记录的耗尽窗口全部到期后，允许新的明确放行观察解除，兼容用量百分比滞后
 pub(super) fn reconcile_refresh(
    current: QuotaState,
    refreshed: &mut CodexAccountQuotaSnapshot,
@@ -139,6 +136,11 @@ pub(super) fn reconcile_refresh(
    document: &mut Map<String, Value>,
 ) -> Result<QuotaState, CodexCredentialQuotaError> {
    document.remove(RECOVERY_FIELD);
+    if refreshed.credits_usable() {
+        // 点数独立于套餐窗口，恢复后不能继续携带已失效的窗口恢复进度
+        refreshed.recovery = None;
+        return Ok(QuotaState::allowed(refreshed.observed_at()));
+    }
    let state = if current.is_exhausted() {
        current
    } else {
@@ -155,7 +157,18 @@ pub(super) fn reconcile_refresh(
    if recovery.pending.is_empty() {
        recovery = QuotaRecovery::new(state, refreshed);
    }
-    if current.is_exhausted() && recovery.observe(refreshed) {
+    // 只信任本次耗尽之后的边界，防止再次拒绝后复用已经过期的 reset 反复解封
+    // 多个耗尽窗口必须全部到期，未知 reset 仍走原有恢复规则
+    let allowed_after_reset = refreshed.quota().access() == QuotaAccessState::Allowed
+        && !recovery.pending.is_empty()
+        && recovery.pending.values().all(|reset| {
+            reset.is_some_and(|reset| {
+                reset.timestamp_micros() > recovery.exhausted_at_micros
+                    && SystemTime::from(reset) <= refreshed.observed_at()
+            })
+        });
+    if current.is_exhausted() && (allowed_after_reset || recovery.observe(refreshed)) {
+        refreshed.recovery = None;
        return Ok(QuotaState::allowed(refreshed.observed_at()));
    }
    let reset_at = recovery
@@ -1,6 +1,7 @@
 //! 请求级额度投影、首查与周期刷新排程

 use super::*;
+use tokio::time::Instant as TokioInstant;

 impl CodexCredentialQuotaService {
    /// 批量预热请求级额度投影；持久层或 Provider JSON 异常只退化为未知额度
@@ -67,7 +68,7 @@ struct CodexQuotaProjectionState {
 }

 struct CodexQuotaRefreshAttempt {
-    monotonic_at: Instant,
+    monotonic_at: TokioInstant,
    wall_at: SystemTime,
 }

@@ -230,7 +231,7 @@ impl CodexQuotaSchedulingProjection {
            .iter()
            .map(|(account, _)| account.id().clone())
            .collect::<BTreeSet<_>>();
-        let refreshed_at = Instant::now();
+        let refreshed_at = TokioInstant::now();
        let mut state = self
            .state
            .write()
@@ -239,12 +240,11 @@ impl CodexQuotaSchedulingProjection {
            .last_periodic_refresh_at
            .retain(|account_id, _| candidate_ids.contains(account_id));

-        // 已耗尽账号首次立即复核，之后每 30 分钟复核，以发现官方提前重置
-        // reset + 2 分钟额外触发一次复核，给上游重置留出传播时间
+        // 已耗尽账号到达 reset 后短周期复核，其余时间保留 30 分钟兜底
        // 正常账号仅在非零用量窗口经过宽限期后参与，未更新时复用周期节流
        let mut reserved = Vec::new();
        for (account, target_reset) in candidates {
-            if !periodic_quota_refresh_due(&state, account.id(), target_reset, now, refreshed_at) {
+            if !periodic_quota_refresh_due(&state, &account, target_reset, now, refreshed_at) {
                continue;
            }
            state.last_periodic_refresh_at.insert(
@@ -364,22 +364,34 @@ fn quota_refresh_candidate(

 fn periodic_quota_refresh_due(
    state: &CodexQuotaProjectionState,
-    account_id: &ProviderAccountId,
+    account: &ProviderAccount,
    reset_at: Option<SystemTime>,
    now: SystemTime,
-    monotonic_now: Instant,
+    monotonic_now: TokioInstant,
 ) -> bool {
-    state
-        .last_periodic_refresh_at
-        .get(account_id)
-        .is_none_or(|last| {
-            monotonic_now.saturating_duration_since(last.monotonic_at)
-                >= PERIODIC_QUOTA_REFRESH_RETRY_INTERVAL
-                || reset_at
-                    .and_then(|reset| reset.checked_add(QUOTA_RESET_GRACE))
-                    // 已在该边界之后复核过时回到周期重试，避免过期 reset 每轮触发
-                    .is_some_and(|due_at| last.wall_at < due_at && due_at <= now)
-        })
+    let mut interval = PERIODIC_QUOTA_REFRESH_RETRY_INTERVAL;
+    let due_at = if account.quota().is_exhausted() {
+        if let Some(since_reset) = reset_at.and_then(|reset| now.duration_since(reset).ok()) {
+            // 上游边界附近可能仍返回旧额度，首次复核也保留传播余量
+            if since_reset < EXHAUSTED_QUOTA_RESET_GRACE {
+                return false;
+            }
+            interval = if since_reset < QUOTA_RESET_FAST_REFRESH_WINDOW {
+                QUOTA_RESET_FAST_REFRESH_INTERVAL
+            } else {
+                QUOTA_RESET_REFRESH_RETRY_INTERVAL
+            };
+        }
+        reset_at.and_then(|reset| reset.checked_add(EXHAUSTED_QUOTA_RESET_GRACE))
+    } else {
+        reset_at.and_then(|reset| reset.checked_add(QUOTA_RESET_GRACE))
+    };
+    let Some(last) = state.last_periodic_refresh_at.get(account.id()) else {
+        return true;
+    };
+    monotonic_now.saturating_duration_since(last.monotonic_at) >= interval
+        // 跨越 reset 宽限期的首轮复核不受旧间隔阻挡，之后按对应间隔节流
+        || due_at.is_some_and(|due_at| last.wall_at < due_at && due_at <= now)
 }

 fn insert_projection_entry(
@@ -515,7 +515,7 @@ impl CodexCredentialQuotaService {
            .await
    }

-    /// 真实限额失败后的异步刷新只补齐展示快照，不允许 usage 快照撤销已确认的失败状态
+    /// 真实限额失败后的异步回读仅允许明确可用的点数恢复访问，套餐状态仍保留失败保护
    pub(crate) async fn refresh_account_after_failure(
        &self,
        account_id: &ProviderAccountId,
@@ -589,9 +589,10 @@ impl CodexCredentialQuotaService {
        } else {
            None
        };
-        if account.quota().is_exhausted()
-            && matches!(authority, QuotaRefreshAuthority::PreserveAccess)
-        {
+        // 失败后回读只允许明确可用的点数恢复访问，套餐 allowed 仍可能滞后
+        let observe_access =
+            matches!(authority, QuotaRefreshAuthority::ObserveAccess) || snapshot.credits_usable();
+        if account.quota().is_exhausted() && !observe_access {
            if self
                .store
                .touch_quota_observation(QuotaObservationTouch {
@@ -615,14 +616,15 @@ impl CodexCredentialQuotaService {
                    .with_observed_at(observed_at),
            });
        }
-        let state = match authority {
-            QuotaRefreshAuthority::ObserveAccess => reconcile_refresh(
+        let state = if observe_access {
+            reconcile_refresh(
                account.quota(),
                &mut snapshot,
                previous.as_ref(),
                &mut object,
-            )?,
-            QuotaRefreshAuthority::PreserveAccess => account.quota(),
+            )?
+        } else {
+            account.quota()
        };
        let snapshot = snapshot.with_quota_state(state);
        if self
@@ -116,6 +116,7 @@ pub struct CodexAccountQuotaSnapshot {
    observed_at: SystemTime,
    plan_type: Option<String>,
    credits: Option<CreditsSnapshot>,
+    credits_usable: bool,
    fact: CodexQuotaFact,
    quota: QuotaState,
    windows: Vec<CodexQuotaWindow>,
@@ -144,12 +145,16 @@ impl CodexAccountQuotaSnapshot {
        self.plan_type.as_deref()
    }

-    /// 点数只用于展示，不改变套餐限额或账号可用性
+    /// 上游明确返回的点数事实；点数可用性独立于套餐窗口
    #[must_use]
    pub fn credits(&self) -> Option<&CreditsSnapshot> {
        self.credits.as_ref()
    }

+    pub(super) const fn credits_usable(&self) -> bool {
+        self.credits_usable
+    }
+
    #[must_use]
    pub const fn fact(&self) -> CodexQuotaFact {
        self.fact
@@ -209,7 +214,7 @@ fn parse_codex_quota_object(
    for limit in canonical_rate_limits(object)? {
        aggregate.observe_rate_limit(limit.rate_limit)?;
    }
-    // credits / spend_control 只证明文档结构已识别，不参与账号级访问结论
+    // 点数与支出控制不换算成套餐剩余百分比，访问结论由账号快照统一判断
    if let Some(spend_control) = object.get("spend_control") {
        aggregate.observe_metadata_object(spend_control, "reached")?;
    }
@@ -345,6 +350,12 @@ pub(crate) fn parse_account_quota_snapshot(
            .then_with(|| left.key.cmp(&right.key))
    });
    let quota = authoritative_quota_state(&object, &windows, observed_at)?;
+    let credits_usable = has_usable_credits(&object);
+    let quota = if credits_usable {
+        QuotaState::allowed(observed_at)
+    } else {
+        quota
+    };
    let recovery = object
        .get(RECOVERY_FIELD)
        .map(|value| serde_json::from_value(value.clone()))
@@ -359,6 +370,7 @@ pub(crate) fn parse_account_quota_snapshot(
            .and_then(Value::as_str)
            .map(str::to_owned),
        credits: object.get("credits").and_then(parse_credits_from_object),
+        credits_usable,
        fact,
        quota,
        windows,
@@ -374,7 +386,27 @@ fn quota_source_order(source: &str) -> u8 {
    }
 }

-/// 默认 `codex` 桶是唯一可改变账号额度访问结论的 quota 响应字段
+fn has_usable_credits(usage: &Map<String, Value>) -> bool {
+    let Some(credits) = usage.get("credits") else {
+        return false;
+    };
+    let available = credits.get("has_credits").and_then(Value::as_bool) == Some(true)
+        || credits.get("unlimited").and_then(Value::as_bool) == Some(true);
+    // 套餐耗尽不阻止点数支出，但不能绕过显式支出上限或工作区限制
+    let spend_reached = usage
+        .get("spend_control")
+        .and_then(|spend| spend.get("reached"))
+        .and_then(Value::as_bool)
+        == Some(true);
+    let other_limit_reached = usage
+        .get("rate_limit_reached_type")
+        .and_then(|reason| reason.get("type"))
+        .and_then(Value::as_str)
+        .is_some_and(|reason| reason != "rate_limit_reached");
+    available && !spend_reached && !other_limit_reached
+}
+
+/// 只有默认 `codex` 桶能提供账号级套餐访问结论，具名桶只保留展示事实
 fn authoritative_quota_state(
    usage: &Map<String, Value>,
    windows: &[CodexQuotaWindow],
@@ -0,0 +1,215 @@
+//! 点数独立可用性、耗尽恢复和过期查询隔离
+
+use super::*;
+
+fn exhausted_usage() -> serde_json::Value {
+    json!({"rate_limit": {
+        "allowed": false,
+        "limit_reached": true,
+        "primary_window": {"used_percent": 100, "reset_at": 1_900_000_000},
+        "secondary_window": {"used_percent": 100, "reset_at": 1_900_604_800}
+    }})
+}
+
+async fn mount_usage(server: &MockServer, usage: serde_json::Value) {
+    server.reset().await;
+    Mock::given(method("GET"))
+        .and(path("/api/codex/usage"))
+        .respond_with(ResponseTemplate::new(200).set_body_json(usage))
+        .mount(server)
+        .await;
+}
+
+#[tokio::test]
+async fn credits_allow_exhausted_subscription_without_bypassing_spend_restrictions() {
+    for (metadata, allowed) in [
+        (
+            json!({"credits": {"has_credits": true, "unlimited": false}}),
+            true,
+        ),
+        (
+            json!({"credits": {"has_credits": false, "unlimited": true}}),
+            true,
+        ),
+        (json!({"credits": {"has_credits": true}}), true),
+        (json!({"credits": {"unlimited": true}}), true),
+        (
+            json!({"credits": {"has_credits": false, "unlimited": false, "balance": "500"}}),
+            false,
+        ),
+        (json!({"credits": null}), false),
+        (
+            json!({"credits": {"has_credits": true}, "spend_control": {"reached": true}}),
+            false,
+        ),
+        (
+            json!({"credits": {"has_credits": true}, "rate_limit_reached_type": {"type": "spend_limit_reached"}}),
+            false,
+        ),
+        (
+            json!({"credits": {"has_credits": true}, "rate_limit_reached_type": {"type": "rate_limit_reached"}}),
+            true,
+        ),
+    ] {
+        let store = Arc::new(MemoryAccountStore::default());
+        create_account_with_enabled(&store, "acct_credits", false).await;
+        let account = store.account("acct_credits").expect("account");
+        persist_credential_state(&store, &account, CredentialState::Expired).await;
+        let server = MockServer::start().await;
+        let service = quota_service_with_base_url(&store, reqwest::Client::new(), server.uri());
+        let mut usage = exhausted_usage();
+        usage
+            .as_object_mut()
+            .expect("usage")
+            .extend(metadata.as_object().expect("metadata").clone());
+        mount_usage(&server, usage).await;
+
+        let snapshot = service
+            .refresh_account(account.id())
+            .await
+            .expect("refresh credits");
+        assert_eq!(
+            snapshot.quota().access() == QuotaAccessState::Allowed,
+            allowed,
+            "{metadata}"
+        );
+        assert_eq!(snapshot.fact().remaining_percent(), Some(0));
+        let current = store.account("acct_credits").expect("current account");
+        assert_eq!(current.quota().is_exhausted(), !allowed, "{metadata}");
+        assert!(!current.enabled());
+        assert_eq!(current.credential_state(), CredentialState::Expired);
+    }
+}
+
+#[tokio::test]
+async fn credits_refresh_clears_old_recovery_and_locks_again_only_when_spent() {
+    for periodic in [false, true] {
+        let store = Arc::new(MemoryAccountStore::default());
+        create_account(&store, "acct_credits_recovery").await;
+        let account = store.account("acct_credits_recovery").expect("account");
+        let server = MockServer::start().await;
+        let service = quota_service_with_base_url(&store, reqwest::Client::new(), server.uri());
+        mount_usage(&server, exhausted_usage()).await;
+        service
+            .refresh_account(account.id())
+            .await
+            .expect("seed exhaustion");
+        assert!(
+            store
+                .quota_json("acct_credits_recovery")
+                .expect("quota")
+                .get("_quota_recovery")
+                .is_some()
+        );
+
+        let mut usage = exhausted_usage();
+        usage["credits"] =
+            json!({"has_credits": !periodic, "unlimited": periodic, "balance": "20"});
+        mount_usage(&server, usage.clone()).await;
+        if periodic {
+            assert_eq!(
+                service
+                    .synchronize()
+                    .await
+                    .expect("periodic recovery")
+                    .updated,
+                1
+            );
+        } else {
+            service
+                .refresh_account(account.id())
+                .await
+                .expect("manual recovery");
+        }
+        let snapshot = service
+            .read_account(account.id())
+            .await
+            .expect("persisted snapshot")
+            .expect("snapshot");
+        assert_eq!(snapshot.quota().access(), QuotaAccessState::Allowed);
+        assert_eq!(snapshot.fact().remaining_percent(), Some(0));
+        assert_eq!(
+            snapshot.credits().expect("credits").balance.as_deref(),
+            Some("20")
+        );
+        assert!(
+            store
+                .quota_json("acct_credits_recovery")
+                .expect("quota")
+                .get("_quota_recovery")
+                .is_none()
+        );
+
+        usage["credits"] = json!({"has_credits": false, "unlimited": false, "balance": "0"});
+        mount_usage(&server, usage).await;
+        let snapshot = service
+            .refresh_account(account.id())
+            .await
+            .expect("spent credits");
+        assert!(snapshot.quota().is_exhausted());
+        assert!(
+            store
+                .account("acct_credits_recovery")
+                .expect("account")
+                .quota()
+                .is_exhausted()
+        );
+    }
+}
+
+#[tokio::test]
+async fn credits_query_started_before_new_exhaustion_cannot_restore_access() {
+    let store = Arc::new(MemoryAccountStore::default());
+    create_account(&store, "acct_stale_credits").await;
+    let account = store.account("acct_stale_credits").expect("account");
+    let server = MockServer::start().await;
+    let service = quota_service_with_base_url(&store, reqwest::Client::new(), server.uri());
+    let mut usage = exhausted_usage();
+    usage["credits"] = json!({"has_credits": true, "unlimited": false});
+    Mock::given(method("GET"))
+        .and(path("/api/codex/usage"))
+        .respond_with(
+            ResponseTemplate::new(200)
+                .set_delay(Duration::from_millis(200))
+                .set_body_json(usage),
+        )
+        .mount(&server)
+        .await;
+    let (result, ()) = tokio::join!(service.refresh_account(account.id()), async {
+        tokio::time::timeout(Duration::from_secs(2), async {
+            while server
+                .received_requests()
+                .await
+                .expect("requests")
+                .is_empty()
+            {
+                tokio::time::sleep(Duration::from_millis(1)).await;
+            }
+        })
+        .await
+        .expect("query started");
+        persist_quota_state(
+            &store,
+            &account,
+            QuotaState::exhausted(QuotaEvidence::UsageLimitReached, SystemTime::now(), None),
+        )
+        .await;
+    });
+    result.expect("old display observation can persist independently of newer access state");
+    assert!(
+        store
+            .account("acct_stale_credits")
+            .expect("account")
+            .quota()
+            .is_exhausted()
+    );
+    assert!(
+        service
+            .read_account(account.id())
+            .await
+            .expect("persisted quota")
+            .expect("snapshot")
+            .quota()
+            .is_exhausted()
+    );
+}
@@ -1,10 +1,12 @@
 //! OpenAI 额度事实边界与展示快照回归

 mod capacity_freeze;
+mod credits;
 mod initial_sync;
 mod recovery;
 mod refresh_timing;
 mod reset_credits;
+mod reset_recovery;
 mod scheduling;
 mod slots;
 mod snapshot;
@@ -1,4 +1,4 @@
-//! 通过 quota 服务验证周期复核与 reset 宽限期，不依赖内部调度状态
+//! 通过 quota 服务验证耗尽复核节奏、恢复与正常窗口宽限期

 use std::time::Duration;

@@ -17,8 +17,17 @@ fn reset_grace(reset: i64) -> SystemTime {
    SystemTime::UNIX_EPOCH + Duration::from_secs((reset + 120) as u64)
 }

+async fn advance_clock(now: &mut SystemTime, seconds: u64) {
+    let duration = Duration::from_secs(seconds);
+    // 只在推进时暂停 Tokio 时钟，HTTP 请求期间保持真实计时
+    tokio::time::pause();
+    tokio::time::advance(duration).await;
+    tokio::time::resume();
+    *now += duration;
+}
+
 #[tokio::test]
-async fn reset_grace_bypasses_periodic_throttle_once_then_allows_the_next_window() {
+async fn exhausted_reset_bypasses_periodic_throttle_and_rechecks_each_pending_window() {
    let mut now = SystemTime::now();
    let store = Arc::new(MemoryAccountStore::default());
    create_account(&store, "acct_quota_timing").await;
@@ -30,8 +39,8 @@ async fn reset_grace_bypasses_periodic_throttle_once_then_allows_the_next_window
        .duration_since(SystemTime::UNIX_EPOCH)
        .unwrap()
        .as_secs() as i64
-        - 115;
-    let week_reset = short_reset + 3;
+        + 5;
+    let week_reset = short_reset + 30;
    let usage = |short_used, short_reset, week_used, week_reset| {
        json!({"rate_limit": {
            "allowed": false,
@@ -55,20 +64,28 @@ async fn reset_grace_bypasses_periodic_throttle_once_then_allows_the_next_window
    let requests = server.received_requests().await.expect("requests").len();

    assert!(
-        now < reset_grace(short_reset),
-        "fixture must precede grace deadline"
+        now < SystemTime::UNIX_EPOCH + Duration::from_secs(short_reset as u64),
+        "fixture must precede reset"
    );
-    service
-        .synchronize_at(now)
-        .await
-        .expect("before grace deadline");
+    service.synchronize_at(now).await.expect("before reset");
    assert_eq!(
        server.received_requests().await.expect("requests").len(),
        requests
    );

    mount_usage(&server, usage(0, short_reset + 18_000, 100, week_reset)).await;
-    now = reset_grace(short_reset);
+    for delay in [0, 9] {
+        now = SystemTime::UNIX_EPOCH + Duration::from_secs(short_reset as u64 + delay);
+        service.synchronize_at(now).await.expect("reset grace");
+        assert!(
+            server
+                .received_requests()
+                .await
+                .expect("requests")
+                .is_empty()
+        );
+    }
+    now = SystemTime::UNIX_EPOCH + Duration::from_secs(short_reset as u64 + 10);
    let partial = service
        .synchronize_at(now)
        .await
@@ -85,11 +102,11 @@ async fn reset_grace_bypasses_periodic_throttle_once_then_allows_the_next_window
    service
        .synchronize_at(now)
        .await
-        .expect("next scan before weekly grace deadline");
+        .expect("next scan before weekly reset");
    assert_eq!(server.received_requests().await.expect("requests").len(), 1);

    // 周窗口的到期复核仍未恢复时，也不能每轮扫描重复请求
-    now = reset_grace(week_reset);
+    now = SystemTime::UNIX_EPOCH + Duration::from_secs(week_reset as u64 + 10);
    assert_eq!(
        service
            .synchronize_at(now)
@@ -127,19 +144,139 @@ async fn periodic_checks_continue_when_reset_is_unknown_or_far_in_the_future() {
            .expect("seed exhaustion");
        mount_usage(&server, usage).await;

+        let mut now = SystemTime::now();
        assert_eq!(
            service
-                .synchronize()
+                .synchronize_at(now)
                .await
                .expect("periodic check")
                .exhausted,
            1
        );
-        service.synchronize().await.expect("throttled repeat check");
+        advance_clock(&mut now, 29 * 60).await;
+        service
+            .synchronize_at(now)
+            .await
+            .expect("throttled repeat check");
        assert_eq!(server.received_requests().await.expect("requests").len(), 1);
+        advance_clock(&mut now, 60).await;
+        service.synchronize_at(now).await.expect("periodic retry");
+        assert_eq!(server.received_requests().await.expect("requests").len(), 2);
    }
 }

+#[tokio::test]
+async fn exhausted_reset_refreshes_quickly_then_backs_off_and_recovers_from_usage() {
+    let store = Arc::new(MemoryAccountStore::default());
+    create_account(&store, "acct_reset_retry").await;
+    let account = store.account("acct_reset_retry").expect("account");
+    let server = MockServer::start().await;
+    let service = quota_service_with_base_url(&store, reqwest::Client::new(), server.uri());
+    let reset = Utc::now().timestamp();
+    let mut now = SystemTime::UNIX_EPOCH + Duration::from_secs(reset as u64);
+    let usage = |allowed, used, reset| {
+        json!({"rate_limit": {
+            "allowed": allowed,
+            "primary_window": {"used_percent": used, "reset_at": reset}
+        }})
+    };
+    mount_usage(&server, usage(false, 100, reset)).await;
+    service
+        .refresh_account(account.id())
+        .await
+        .expect("seed exhaustion");
+    service
+        .synchronize_at(now)
+        .await
+        .expect("reset grace without periodic history");
+    advance_clock(&mut now, 9).await;
+    service
+        .synchronize_at(now)
+        .await
+        .expect("still in reset grace");
+    assert_eq!(server.received_requests().await.expect("requests").len(), 1);
+    advance_clock(&mut now, 1).await;
+    service
+        .synchronize_at(now)
+        .await
+        .expect("initial reset check");
+    assert_eq!(server.received_requests().await.expect("requests").len(), 2);
+
+    // 本次拒绝发生在 reset 之后，滞后的 allowed 不能复用旧边界解封
+    mount_usage(&server, usage(true, 100, reset)).await;
+    for (elapsed, requests) in [
+        (29, 0),
+        (1, 1),
+        (29, 1),
+        (1, 2),
+        (229, 3),
+        (1, 3),
+        (298, 3),
+        (1, 4),
+    ] {
+        advance_clock(&mut now, elapsed).await;
+        service.synchronize_at(now).await.expect("scheduled check");
+        assert_eq!(
+            server.received_requests().await.expect("requests").len(),
+            requests
+        );
+        assert!(
+            store
+                .account("acct_reset_retry")
+                .expect("account")
+                .quota()
+                .is_exhausted()
+        );
+    }
+
+    // 查询失败也占用一次尝试，避免网络故障时取消节流
+    server.reset().await;
+    Mock::given(method("GET"))
+        .and(path("/api/codex/usage"))
+        .respond_with(ResponseTemplate::new(429))
+        .mount(&server)
+        .await;
+    advance_clock(&mut now, 300).await;
+    assert_eq!(
+        service
+            .synchronize_at(now)
+            .await
+            .expect("failed check")
+            .transient,
+        1
+    );
+    advance_clock(&mut now, 299).await;
+    service
+        .synchronize_at(now)
+        .await
+        .expect("throttled failed check");
+    assert_eq!(server.received_requests().await.expect("requests").len(), 1);
+
+    mount_usage(&server, usage(true, 0, reset + 18_000)).await;
+    advance_clock(&mut now, 1).await;
+    assert_eq!(
+        service
+            .synchronize_at(now)
+            .await
+            .expect("recovery check")
+            .updated,
+        1
+    );
+    assert!(
+        !store
+            .account("acct_reset_retry")
+            .expect("account")
+            .quota()
+            .is_exhausted()
+    );
+    advance_clock(&mut now, 300).await;
+    service
+        .synchronize_at(now)
+        .await
+        .expect("recovered account leaves retry schedule");
+    assert_eq!(server.received_requests().await.expect("requests").len(), 1);
+}
+
 #[tokio::test]
 async fn allowed_account_with_expired_window_synchronizes_at_reset_grace() {
    let mut now = SystemTime::now();
@@ -0,0 +1,160 @@
+//! 到期后的明确放行恢复，保留到期前、多窗口与再次拒绝的保护
+
+use gateway_core::account::{OpaqueProviderData, QuotaObservation};
+
+use super::*;
+
+fn usage(now: i64, resets: &[Option<i64>], allowed: Option<bool>) -> serde_json::Value {
+    let mut value = json!({"rate_limit": {}});
+    if let Some(allowed) = allowed {
+        value["rate_limit"]["allowed"] = json!(allowed);
+    }
+    for ((role, seconds), reset) in [("primary_window", 18_000), ("secondary_window", 604_800)]
+        .into_iter()
+        .zip(resets)
+    {
+        let mut window = json!({"used_percent": 100, "limit_window_seconds": seconds});
+        if let Some(offset) = reset {
+            window["reset_at"] = json!(now + offset);
+        }
+        value["rate_limit"][role] = window;
+    }
+    value
+}
+
+#[tokio::test]
+async fn allowed_refresh_recovers_full_windows_only_after_the_confirmed_resets() {
+    for (name, resets, failure_offset, allowed, recovered) in [
+        ("before_reset", vec![Some(3600)], -120, Some(true), false),
+        ("after_reset", vec![Some(-60)], -120, Some(true), true),
+        ("still_denied", vec![Some(-60)], -120, Some(false), false),
+        ("unknown_access", vec![Some(-60)], -120, None, false),
+        ("unknown_reset", vec![None], -120, Some(true), false),
+        ("new_failure", vec![Some(-60)], -30, Some(true), false),
+        (
+            "weekly_not_due",
+            vec![Some(-60), Some(3600)],
+            -120,
+            Some(true),
+            false,
+        ),
+        (
+            "both_due",
+            vec![Some(-90), Some(-60)],
+            -120,
+            Some(true),
+            true,
+        ),
+        (
+            "weekly_unknown",
+            vec![Some(-60), None],
+            -120,
+            Some(true),
+            false,
+        ),
+    ] {
+        for worker in [false, true] {
+            let store = Arc::new(MemoryAccountStore::default());
+            let account_id = format!("acct_reset_{name}_{worker}");
+            create_account(&store, &account_id).await;
+            let account = store.account(&account_id).expect("account");
+            let now = Utc::now().timestamp();
+            let observed_at =
+                SystemTime::UNIX_EPOCH + Duration::from_secs((now + failure_offset) as u64);
+            let reset_at = resets
+                .iter()
+                .filter_map(|reset| *reset)
+                .min()
+                .map(|offset| SystemTime::UNIX_EPOCH + Duration::from_secs((now + offset) as u64));
+            let old_usage = usage(now, &resets, Some(false));
+            store
+                .compare_and_swap_quota(QuotaObservation {
+                    plan_type: None,
+                    account_id: account.id().clone(),
+                    expected_revision: account.revision(),
+                    quota: OpaqueProviderData::new(old_usage.as_object().expect("usage").clone()),
+                    observed_at,
+                    state: QuotaState::exhausted(
+                        QuotaEvidence::UsageLimitReached,
+                        observed_at,
+                        reset_at,
+                    ),
+                })
+                .await
+                .expect("persist confirmed exhaustion before refresh");
+            let server = MockServer::start().await;
+            Mock::given(method("GET"))
+                .and(path("/api/codex/usage"))
+                .respond_with(
+                    ResponseTemplate::new(200).set_body_json(usage(now, &resets, allowed)),
+                )
+                .mount(&server)
+                .await;
+            let service = quota_service_with_base_url(&store, reqwest::Client::new(), server.uri());
+            if worker {
+                let summary = service.synchronize().await.expect("worker refresh");
+                assert_eq!(summary.updated, u64::from(recovered), "{name}");
+            } else {
+                service
+                    .refresh_account(account.id())
+                    .await
+                    .expect("manual refresh");
+            }
+            let refreshed = service
+                .read_account(account.id())
+                .await
+                .expect("read")
+                .expect("snapshot");
+            assert_eq!(
+                refreshed.quota().is_exhausted(),
+                !recovered,
+                "{name}, worker={worker}"
+            );
+            assert!(
+                refreshed
+                    .windows()
+                    .iter()
+                    .all(|window| window.used_percent() == Some(100.0))
+            );
+            assert_eq!(
+                store
+                    .account(&account_id)
+                    .expect("account")
+                    .quota()
+                    .is_exhausted(),
+                !recovered
+            );
+            assert_eq!(
+                store
+                    .quota_json(&account_id)
+                    .expect("quota")
+                    .get("_quota_recovery")
+                    .is_none(),
+                recovered
+            );
+
+            if recovered {
+                service
+                    .refresh_account(account.id())
+                    .await
+                    .expect("allowed remains allowed");
+                assert!(
+                    !store
+                        .account(&account_id)
+                        .expect("account")
+                        .quota()
+                        .is_exhausted()
+                );
+                // 新拒绝不能沿用刚刚解除时的旧 reset，即使查询仍然 allowed=true
+                persist_quota_state(&store, &account, exhausted_quota(reset_at)).await;
+                for _ in 0..2 {
+                    let snapshot = service
+                        .refresh_account(account.id())
+                        .await
+                        .expect("refresh after new refusal");
+                    assert!(snapshot.quota().is_exhausted(), "{name}");
+                }
+            }
+        }
+    }
+}
@@ -8421,6 +8421,82 @@ async fn official_usage_limit_failure_persists_fact_without_fabricating_usage()
    );
 }

+#[tokio::test]
+async fn usage_limit_failure_recovers_with_credits_without_replaying_the_failed_request() {
+    let store = Arc::new(MemoryAccountStore::default());
+    let account_id = "acct_provider_contract";
+    create_account(&store, account_id).await;
+    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": true,
+                "primary_window": {"used_percent": 100, "reset_at": 1_900_000_000}
+            },
+            "credits": {"has_credits": true, "unlimited": false, "balance": "20"}
+        })))
+        .expect(1)
+        .mount(&server)
+        .await;
+    Mock::given(method("POST"))
+        .and(path("/codex/responses"))
+        .respond_with(ResponseTemplate::new(429).set_body_json(json!({
+            "error": {"type": "usage_limit_reached", "message": "usage limit reached", "resets_at": 1_900_000_000}
+        })))
+        .expect(1)
+        .mount(&server)
+        .await;
+    let mut stream = provider_with_base_url(&store, server.uri())
+        .execute(
+            planned_request("openai", http_generate_operation()),
+            context("req_usage_limit_credits", CancellationToken::new()),
+        )
+        .await
+        .expect("provider stream");
+    let failure = loop {
+        match stream.next().await {
+            Some(Ok(_)) => {}
+            Some(Err(error)) => break error,
+            None => panic!("quota failure must reach the caller"),
+        }
+    };
+    assert_eq!(failure.kind(), ProviderErrorKind::QuotaExhausted);
+    assert!(
+        store
+            .account(account_id)
+            .expect("account")
+            .quota()
+            .is_exhausted()
+    );
+
+    timeout(Duration::from_secs(5), async {
+        while store
+            .account(account_id)
+            .expect("account")
+            .quota()
+            .is_exhausted()
+        {
+            tokio::time::sleep(Duration::from_millis(10)).await;
+        }
+    })
+    .await
+    .expect("credits restore access after background refresh");
+    assert!(stream.next().await.is_none());
+    let account = store.account(account_id).expect("recovered account");
+    assert_eq!(account.quota().access(), QuotaAccessState::Allowed);
+    assert_eq!(account.credential_state(), CredentialState::Ready);
+    let quota = store.quota_json(account_id).expect("persisted credits");
+    assert_eq!(quota.pointer("/credits/has_credits"), Some(&json!(true)));
+    assert_eq!(
+        quota.pointer("/rate_limits_by_limit_id/codex/primary_window/used_percent"),
+        Some(&json!(100))
+    );
+    assert!(quota.get("_quota_recovery").is_none());
+    server.verify().await;
+}
+
 #[tokio::test]
 async fn capacity_http_and_websocket_opening_rejections_use_bounded_business_retry() {
    for use_websocket in [false, true] {
@@ -430,7 +430,7 @@ impl ProviderAccountStore for MemoryAccountStore {

    async fn compare_and_swap_quota(
        &self,
-        observation: QuotaObservation,
+        mut observation: QuotaObservation,
    ) -> Result<QuotaWriteOutcome, StoreError> {
        let mut accounts = self.accounts.lock().expect("account store lock");
        let stored = accounts
@@ -446,7 +446,18 @@ impl ProviderAccountStore for MemoryAccountStore {
        {
            return Ok(QuotaWriteOutcome::Conflict);
        }
-        let quota = observation.state;
+        // 与 PostgreSQL 一致，展示快照与访问事实分别按各自观察时间隔离
+        let current = stored.account.quota();
+        let quota = if observation
+            .state
+            .observed_at()
+            .is_some_and(|observed| current.observed_at().is_none_or(|time| time <= observed))
+        {
+            observation.state
+        } else {
+            current
+        };
+        observation.state = quota;
        let mut replacement = AccountRebuild::preserving(&stored.account).with_quota(quota);
        if let Some(plan_type) = &observation.plan_type {
            replacement.profile = Some((
@@ -754,7 +754,10 @@ Codex 的 `model_catalog_json` 配置。账号设置保存不等待上游模型
 `quota.credits` 返回上游点数余额，未提供点数信息时为 `null`。对象包含 `hasCredits`、`unlimited` 和
 `balance`，余额为保留上游精度的十进制字符串，未提供余额时为 `null`，明确的零余额为 `"0"`。
 OpenAI 点数随现有额度刷新和正常请求的额度信息同步，不与主动重置卡库存或百分比限额合并，
-不参与账号可用性判断
+不根据 `balance` 单独推导可用性。主动查询明确 `hasCredits=true` 或 `unlimited=true`，且无阻止
+点数支出的上限或工作区限制时，即使套餐窗口耗尽也可使用点数，并解除已有额度耗尽状态。
+真实额度拒绝后的后台回读同样支持此恢复路径，但不会自动重发本次失败请求；点数不能清除
+凭据错误、停用或冷却状态。上游点数信息也可能延迟，后续真实拒绝仍会记录额度耗尽

 账号列表和详情返回 `notes`（无备注时为 `null`）。编辑时省略或 `null` 保留原备注；字符串最多 500 个 Unicode
 字符，允许换行和制表符，保存时去除首尾空白，空字符串清空备注。备注独立于上游身份，导入时未显式提供备注、
@@ -1118,11 +1121,14 @@ OAuth start 使用：
  但上游 authorization code 本身通常只能交换一次；已完成过 token exchange 时应重新创建 OAuth flow
 - `GET /accounts/quota` 只读取最后一次落库快照；`POST /accounts/quota/refresh` 才访问上游。access token
  已过期时，额度刷新要求先走 credential 刷新或重新授权，不会拿过期 token 探测额度
- OpenAI 已耗尽账号每 30 分钟主动复核一次，也会在最早未恢复窗口的 `resetAt + 2 分钟` 到期后
-  提前复核；周期复核不等待旧重置时间，因此也能发现官方提前重置。正常账号有非零用量或触顶窗口时，
-  在该窗口的 `resetAt + 2 分钟` 后主动复核。后台每 30 秒检查触发条件；同一重置边界复核后仍未更新时
-  回到 30 分钟重试，避免旧 reset 持续触发请求。各窗口独立确认恢复，时间到期本身不会直接解除账号
-  耗尽或将展示用量归零
+- OpenAI 已耗尽账号首次可复核，在 `resetAt` 尚未到达或未知时按 30 分钟间隔兜底，也可发现
+  官方提前重置。到达最早未恢复窗口的 `resetAt` 后，先等待 10 秒传播余量，再在下一轮后台检查
+  复核；reset 后前 5 分钟最小查询间隔为 30 秒，之后仍耗尽则改为 5 分钟。正常账号有非零用量或触顶窗口时，仍在
+  `resetAt + 2 分钟` 后复核，未更新则按 30 分钟间隔重试。后台每 30 秒检查条件，实际恢复时间还受
+  排队、查询耗时和上游传播影响；查询失败也遵守间隔。无可用点数时，到期前仍按原窗口证据恢复；
+  本次耗尽之后的已记录窗口 reset 全部到期后，新查询明确 `allowed=true` 可解除限制，即使用量仍为
+  100%。未知 reset 或再次拒绝后残留的旧 reset 仍走原规则，不复用旧边界解封；时间到期本身不会
+  解除限制，也不会将展示用量归零
 - 账号 `enabled` 控制是否参与请求调度，不控制 OAuth 凭据续期。OpenAI 与 xAI 的停用账号仍按各自刷新
  策略维护 Token；刷新结果更新凭据状态，但不会启用调度。无 refresh token 或凭据已进入失效、无效、
  封禁状态的账号不参加后台自动续期
@@ -890,6 +890,12 @@ credential 与 quota 是两组独立事实：credential refresh 不等于 quota
  OpenAI Provider 将其中明确的套餐变更与额度原子提交，共用凭据版本和观察时间保护；
  空值及 `unknown` 不覆盖套餐，同族泛化值保留具体子类型。Token 刷新保留提交时的账号资料

+OpenAI Provider 的额度快照统一解释点数可用性，主动查询、耗尽恢复与失败后回读复用同一规则。
+可用点数独立于套餐窗口；通过点数恢复时原子保存新快照、访问事实并清理旧窗口恢复进度。
+无可用点数的失败回读保留已确认的拒绝；常规刷新在到期前仍按窗口证据恢复，本次耗尽之后的
+已记录窗口 reset 全部到期后可依据新的明确放行观察恢复。查询沿用 credential revision 与
+观察时间隔离，不允许旧查询撤销较新的失败；恢复条件和后台复核节奏见 [账号 API](api.md#账号能力导入与-oauth)
+
 OpenAI 订阅周期属于按需个人信息，不是额度事实。Admin 账号用例通过现有 Provider 管理端口并发读取
 个人资料统计与订阅，汇聚为一次只读响应；Provider 继续拥有各自的认证、出站代理与上游协议处理。
 订阅不由 quota、导入或后台任务触发，不持久化。汇聚结果返回前核对账号身份与 credential revision，