chore: bump version to 3.0.30

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

.github/workflows/release.yml

@@ -626,7 +626,7 @@ jobs:
           make_latest: ${{ needs.release-info.outputs.prerelease == 'true' && 'false' || 'true' }}
           body: |
             > [!WARNING]
-            > **Upgrading across major versions?** Database schema and configuration are only compatible within a major version, and the one-click self-update refuses cross-major targets (releases before 3.0 lack this gate — do not use their built-in update to reach a newer major). Deploy the new major fresh (new `.runtime/` data directory) and re-import or re-authorize provider accounts and client keys. See the [upgrade notes](https://github.com/${{ github.repository }}/blob/main/deploy/README.md#镜像升级与源码构建).
+            > **Upgrading across major versions?** Online upgrades across major versions are not supported. Deploy the new major fresh with a new `.runtime/` data directory, then re-import or re-authorize provider accounts and client keys. See the [upgrade notes](https://github.com/${{ github.repository }}/blob/main/deploy/README.md#镜像升级与源码构建).
 
             ## 📥 Installation
 

README.md

@@ -168,10 +168,8 @@ docker compose -f deploy/compose.yaml up -d
 > 运行状态。
 
 > [!WARNING]
-> 跨大版本不是原地升级：数据库 schema 与配置只在同一大版本内兼容，内置一键自更新
-> 会拒绝跨大版本目标（3.0 之前的旧版没有这道门禁，不要用旧版界面的一键更新跨大版本）。
-> 升级到新的大版本需要全新部署（全新 `.runtime/` 数据目录），并重新导入或重新授权
-> Provider 账号与客户端 Key。
+> 不支持跨大版本在线升级。升级到新的大版本时，请使用全新的 `.runtime/` 数据目录重新部署，
+> 并重新导入或重新授权 Provider 账号与客户端 Key。
 
 Compose 默认只绑定 `127.0.0.1`。公网接入应使用 HTTPS 反向代理，转发 WebSocket
 upgrade 与真实客户端 IP；不要暴露 PostgreSQL 或 Redis。

backend/crates/gateway-admin/src/model/observability.rs

@@ -461,7 +461,9 @@ pub struct RequestMetricPoint {
 
 /// 账号池容量统计。
 ///
-/// 配额耗尽单列为可恢复状态；`unavailable` 只包含过期、禁用和封禁账号。
+/// 首页概览的 `unavailable` 表示不可参与调度，包含配额耗尽及过期、禁用和封禁账号。
+///
+/// `quota_exhausted` 仍单列，以便状态详情展示和诊断。
 #[derive(Debug, Clone, Default, PartialEq, Eq)]
 pub struct AccountPoolMetrics {
     pub total: u64,

backend/crates/gateway-api/src/admin/observability.rs

@@ -2234,9 +2234,6 @@ fn dashboard_view(result: domain::DashboardResult, kind: TrendKind) -> Dashboard
         account_usage,
         recent_requests,
     } = observation;
-    // 概览卡只有可用和不可用两桶：额度耗尽可恢复，保留在可用；不可用只含终态账号。
-    let available_accounts =
-        dashboard_available_account_count(provider_accounts.total, provider_accounts.unavailable);
     let mut account_usage_views = Vec::with_capacity(account_usage.len());
     let mut credential_usage_views = Vec::with_capacity(account_usage.len());
     for credential in account_usage {
@@ -2285,8 +2282,8 @@ fn dashboard_view(result: domain::DashboardResult, kind: TrendKind) -> Dashboard
             credentials: DashboardCredentialsCardView {
                 total: format_compact_number(provider_accounts.total),
                 total_value: provider_accounts.total,
-                available: format_compact_number(available_accounts),
-                available_value: available_accounts,
+                available: format_compact_number(provider_accounts.active),
+                available_value: provider_accounts.active,
                 unavailable: format_compact_number(provider_accounts.unavailable),
                 unavailable_value: provider_accounts.unavailable,
             },
@@ -2342,10 +2339,6 @@ fn dashboard_view(result: domain::DashboardResult, kind: TrendKind) -> Dashboard
     }
 }
 
-fn dashboard_available_account_count(total: u64, unavailable: u64) -> u64 {
-    total.saturating_sub(unavailable)
-}
-
 fn usage_summary_view(summary: domain::UsageSummary) -> UsageSummaryView {
     let overview = summary.overview;
     UsageSummaryView {

backend/crates/gateway-api/tests/admin/observability.rs

@@ -261,7 +261,7 @@ mod response {
 }
 
 #[tokio::test]
-async fn dashboard_summary_should_count_quota_exhausted_accounts_as_available_in_headline_card() {
+async fn dashboard_summary_should_include_quota_exhaustion_in_unavailable_headline_count() {
     use axum::{
         body::{Body, to_bytes},
         http::{Request, StatusCode, header},
@@ -288,7 +288,7 @@ async fn dashboard_summary_should_count_quota_exhausted_accounts_as_available_in
         provider_accounts: AccountPoolMetrics {
             total: 807,
             enabled: 804,
-            unavailable: 7,
+            unavailable: 792,
             active: 15,
             expired: 2,
             quota_exhausted: 785,
@@ -328,8 +328,8 @@ async fn dashboard_summary_should_count_quota_exhausted_accounts_as_available_in
         ),
         (
             &serde_json::json!(807),
-            &serde_json::json!(800),
-            &serde_json::json!(7)
+            &serde_json::json!(15),
+            &serde_json::json!(792)
         ),
     );
 }

backend/crates/gateway-store/src/postgres/observability.rs

@@ -2013,8 +2013,7 @@ async fn provider_account_metrics(
          )
          select count(*)::bigint as total,
                 count(*) filter (where enabled)::bigint as enabled,
-                count(*) filter (where status in ('expired', 'disabled', 'banned'))::bigint
-                  as unavailable,
+                count(*) filter (where status <> 'active')::bigint as unavailable,
                 count(*) filter (where status = 'active')::bigint as active,
                 count(*) filter (where status = 'expired')::bigint as expired,
                 count(*) filter (where status = 'quota_exhausted')::bigint as quota_exhausted,

backend/crates/gateway-store/src/postgres/provider_accounts.rs

@@ -630,7 +630,8 @@ impl ProviderAccountRepository for PgProviderAccountRepository {
                      else availability_observed_at
                  end,
                  updated_at = case when enabled then greatest(now(), $6) else updated_at end
-             where id = $1 and credential_revision = $2",
+             where id = $1 and credential_revision = $2
+               and (availability_observed_at is null or availability_observed_at <= $6)",
         )
         .bind(update.account_id)
         .bind(to_i64(update.expected_revision.get())?)
@@ -675,7 +676,8 @@ impl ProviderAccountRepository for PgProviderAccountRepository {
             "update provider_accounts
              set provider_quota_json = $3, quota_observed_at = $4,
                  updated_at = greatest(now(), coalesce($4, now()))
-             where id = $1 and credential_revision = $2",
+             where id = $1 and credential_revision = $2
+               and ($4 is null or quota_observed_at is null or quota_observed_at <= $4)",
         )
         .bind(account_id)
         .bind(to_i64(expected_revision.get())?)

backend/crates/gateway-store/tests/postgres/mod.rs

@@ -58,30 +58,6 @@ impl TestDatabase {
             .execute(&mut *migration)
             .await
             .expect("apply terminal migration");
-        sqlx::raw_sql(include_str!(
-            "../../../../migrations/0002_drop_opaque_response_id_index.sql"
-        ))
-        .execute(&mut *migration)
-        .await
-        .expect("apply opaque response ID migration");
-        sqlx::raw_sql(include_str!(
-            "../../../../migrations/0003_store_opaque_response_ids_as_bytes.sql"
-        ))
-        .execute(&mut *migration)
-        .await
-        .expect("apply opaque response storage migration");
-        sqlx::raw_sql(include_str!(
-            "../../../../migrations/0004_add_model_request_service_tier.sql"
-        ))
-        .execute(&mut *migration)
-        .await
-        .expect("apply model request service tier migration");
-        sqlx::raw_sql(include_str!(
-            "../../../../migrations/0005_allow_quota_exhausted_cooldown.sql"
-        ))
-        .execute(&mut *migration)
-        .await
-        .expect("apply quota exhausted cooldown migration");
         migration.commit().await.expect("commit terminal migration");
         Some(Self {
             admin,
@@ -177,25 +153,19 @@ async fn connect_and_migrate_should_apply_all_migrations_once_and_reopen_cleanly
     .expect("drop migration test database");
     admin.close().await;
 
-    assert_eq!((first_table_count, migration_count), (8, 5));
+    assert_eq!((first_table_count, migration_count), (8, 1));
     assert_eq!(response_id_types, ["bytea", "bytea"]);
     assert!(!raw_response_id_index_exists);
 }
 
 #[test]
 fn migrations_should_leave_transaction_ownership_to_sqlx() {
-    let transaction_statements = [
-        include_str!("../../../../migrations/0001_initial.sql"),
-        include_str!("../../../../migrations/0002_drop_opaque_response_id_index.sql"),
-        include_str!("../../../../migrations/0003_store_opaque_response_ids_as_bytes.sql"),
-        include_str!("../../../../migrations/0004_add_model_request_service_tier.sql"),
-        include_str!("../../../../migrations/0005_allow_quota_exhausted_cooldown.sql"),
-    ]
-    .into_iter()
-    .flat_map(str::lines)
-    .map(str::trim)
-    .filter(|line| matches!(*line, "begin;" | "commit;"))
-    .count();
+    let transaction_statements = [include_str!("../../../../migrations/0001_initial.sql")]
+        .into_iter()
+        .flat_map(str::lines)
+        .map(str::trim)
+        .filter(|line| matches!(*line, "begin;" | "commit;"))
+        .count();
 
     assert_eq!(transaction_statements, 0);
 }

backend/crates/gateway-store/tests/postgres/observability.rs

@@ -169,12 +169,14 @@ async fn dashboard_account_metrics_should_partition_account_statuses() {
             metrics.disabled,
             metrics.banned,
         ),
-        (6, 5, 2, 3, 1, 1, 1, 1),
+        (6, 5, 2, 4, 1, 1, 1, 1),
     );
+    // 首页按调度可用性聚合：配额耗尽也属于不可用。
     assert_eq!(
-        metrics.total,
-        metrics.active + metrics.quota_exhausted + metrics.unavailable
+        metrics.unavailable,
+        metrics.quota_exhausted + metrics.expired + metrics.disabled + metrics.banned
     );
+    assert_eq!(metrics.total, metrics.active + metrics.unavailable);
     database.close().await;
 }
 

backend/crates/gateway-store/tests/postgres/provider_accounts.rs

@@ -111,6 +111,97 @@ async fn core_quota_batch_reads_only_observed_accounts_in_one_contract_call() {
     database.close().await;
 }
 
+#[tokio::test]
+async fn delayed_provider_observations_cannot_overwrite_newer_state_or_quota() {
+    let Some(database) = TestDatabase::create("provider_account_observation_fence").await else {
+        return;
+    };
+    let repository = PgProviderAccountRepository::new(database.pool.clone());
+    let account_id = ProviderAccountId::new("acct_observation_fence").expect("account id");
+    repository
+        .insert_provider_account(account(account_id.as_str(), "user-observation-fence"))
+        .await
+        .expect("insert observation fence fixture");
+    let revision = CredentialRevision::new(1).expect("revision");
+    let newer = SystemTime::now()
+        .checked_add(Duration::from_secs(60))
+        .expect("newer observation time");
+    let older = newer
+        .checked_sub(Duration::from_secs(30))
+        .expect("older observation time");
+
+    let quota = |marker: &str, observed_at| QuotaObservation {
+        account_id: account_id.clone(),
+        expected_revision: revision,
+        quota: Some(OpaqueProviderData::new(
+            json!({"marker": marker})
+                .as_object()
+                .expect("quota object")
+                .clone(),
+        )),
+        observed_at: Some(observed_at),
+    };
+    assert_eq!(
+        repository
+            .compare_and_swap_quota(quota("newer", newer))
+            .await
+            .expect("persist newer quota"),
+        QuotaWriteOutcome::Updated
+    );
+    assert_eq!(
+        repository
+            .compare_and_swap_quota(quota("older", older))
+            .await
+            .expect("reject older quota"),
+        QuotaWriteOutcome::Conflict
+    );
+
+    repository
+        .apply_state_change(AccountStateChange {
+            account_id: account_id.clone(),
+            expected_revision: revision,
+            availability: AccountAvailability::QuotaExhausted,
+            reason: Some("quota_exhausted".to_owned()),
+            cooldown_until: Some(newer),
+            observed_at: newer,
+        })
+        .await
+        .expect("persist newer account state");
+    assert!(
+        repository
+            .apply_state_change(AccountStateChange {
+                account_id: account_id.clone(),
+                expected_revision: revision,
+                availability: AccountAvailability::Ready,
+                reason: None,
+                cooldown_until: None,
+                observed_at: older,
+            })
+            .await
+            .is_err(),
+        "a delayed state observation must conflict"
+    );
+
+    let current = repository
+        .get_account(&account_id)
+        .await
+        .expect("load fenced account")
+        .expect("fenced account");
+    assert_eq!(current.availability(), AccountAvailability::QuotaExhausted);
+    let observed = repository
+        .get_quotas(std::slice::from_ref(&account_id))
+        .await
+        .expect("load fenced quota")
+        .pop()
+        .expect("fenced quota");
+    assert_eq!(
+        observed.quota.expect("quota document").expose_to_provider()["marker"],
+        "newer"
+    );
+
+    database.close().await;
+}
+
 #[tokio::test]
 async fn terminal_admin_list_filters_and_sorts_before_pagination_with_retained_usage() {
     let Some(database) = TestDatabase::create("provider_account_terminal_list").await else {

backend/crates/providers/openai/src/admin.rs

@@ -25,8 +25,9 @@ use gateway_admin::model::{MutationActor, MutationContext, Revision};
 use gateway_admin::ports::provider::{ProviderAdmin, ProviderAdminError, ProviderAdminErrorKind};
 use gateway_core::accounting::Money;
 use gateway_core::engine::credential::{
-    CredentialRevision, LoadedCredential, NewProviderAccount, OpaqueProviderData,
-    PlaintextCredential, ProviderAccount, ProviderAccountId, ProviderAccountStore,
+    AccountAvailability, CredentialRevision, LoadedCredential, NewProviderAccount,
+    OpaqueProviderData, PlaintextCredential, ProviderAccount, ProviderAccountId,
+    ProviderAccountStore,
 };
 use gateway_core::error::StoreErrorKind;
 use gateway_core::operation::{GenerateRequest, Operation, ProtocolPayload};
@@ -458,21 +459,23 @@ impl ProviderAdmin for OpenAiAdminProvider {
             refresh,
             rolling_usage: _,
         } = request;
-        self.account(&account_id).await?;
+        let mut account = self.account(&account_id).await?;
         let snapshot = if refresh {
-            Some(
+            let snapshot = Some(
                 self.quota
                     .refresh_account(&account_id)
                     .await
                     .map_err(map_quota_error)?,
-            )
+            );
+            account = self.account(&account_id).await?;
+            snapshot
         } else {
             self.quota
                 .read_account(&account_id)
                 .await
                 .map_err(map_quota_error)?
         };
-        Ok(snapshot.map_or_else(empty_quota, project_quota))
+        Ok(project_quota(snapshot, &account))
     }
 
     async fn models(
@@ -746,7 +749,18 @@ fn empty_quota() -> ProviderQuota {
     }
 }
 
-fn project_quota(snapshot: CodexAccountQuotaSnapshot) -> ProviderQuota {
+fn project_quota(
+    snapshot: Option<CodexAccountQuotaSnapshot>,
+    account: &ProviderAccount,
+) -> ProviderQuota {
+    let mut quota = snapshot.map_or_else(empty_quota, project_quota_snapshot);
+    if account.availability() == AccountAvailability::QuotaExhausted {
+        force_confirmed_exhaustion_projection(&mut quota, account.cooldown_until());
+    }
+    quota
+}
+
+fn project_quota_snapshot(snapshot: CodexAccountQuotaSnapshot) -> ProviderQuota {
     let mut provider_data = Map::new();
     provider_data.insert(
         "remaining_percent".to_owned(),
@@ -799,6 +813,68 @@ fn project_quota(snapshot: CodexAccountQuotaSnapshot) -> ProviderQuota {
     }
 }
 
+fn force_confirmed_exhaustion_projection(
+    quota: &mut ProviderQuota,
+    cooldown_until: Option<SystemTime>,
+) {
+    // 真实 429 对“当前已耗尽”的证明强于可能延迟结算的 usage 文档。这里只覆盖
+    // Admin 展示投影，原始 Provider JSON 仍作为后续判断窗口是否真正恢复的基线。
+    quota.provider_data = Some(ProviderDocument::new(OpaqueProviderData::new(
+        Map::from_iter([
+            (
+                "remaining_percent".to_owned(),
+                Value::Number(Number::from(0)),
+            ),
+            ("exhausted".to_owned(), Value::Bool(true)),
+        ]),
+    )));
+    let reset_at = cooldown_until.map(DateTime::<Utc>::from);
+    if quota.windows.is_empty() {
+        quota.windows.push(ProviderQuotaWindow {
+            key: "confirmed-quota-exhaustion".to_owned(),
+            group: "other".to_owned(),
+            label: "额度".to_owned(),
+            source: None,
+            window_seconds: None,
+            used_percent: Some(100.0),
+            reset_at,
+            local_usage: None,
+            provider_data: None,
+        });
+        return;
+    }
+
+    let matched_reset = reset_at.and_then(|reset_at| {
+        quota
+            .windows
+            .iter()
+            .enumerate()
+            .filter_map(|(index, window)| {
+                let distance = window.reset_at.map(|window_reset| {
+                    (window_reset.timestamp() - reset_at.timestamp()).unsigned_abs()
+                })?;
+                Some((index, distance))
+            })
+            .min_by_key(|(_, distance)| *distance)
+            .map(|(index, _)| index)
+    });
+    let index = matched_reset
+        .or_else(|| {
+            quota
+                .windows
+                .iter()
+                .enumerate()
+                .max_by(|(_, left), (_, right)| {
+                    left.used_percent
+                        .unwrap_or(-1.0)
+                        .total_cmp(&right.used_percent.unwrap_or(-1.0))
+                })
+                .map(|(index, _)| index)
+        })
+        .unwrap_or(0);
+    quota.windows[index].used_percent = Some(100.0);
+}
+
 const fn quota_group(kind: CodexQuotaWindowKind) -> &'static str {
     match kind {
         CodexQuotaWindowKind::Monthly => "monthly",

backend/crates/providers/openai/src/config.rs

@@ -217,7 +217,7 @@ pub struct CodexQuotaSettings {
 impl Default for CodexQuotaSettings {
     fn default() -> Self {
         Self {
-            refresh_interval_minutes: 5,
+            refresh_interval_minutes: 15,
             skip_exhausted: true,
         }
     }

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

@@ -30,15 +30,12 @@ const MAX_RATE_LIMIT_COOLDOWN: Duration = Duration::from_secs(60 * 60);
 const QUOTA_SCHEDULING_TTL: Duration = Duration::from_secs(10 * 60);
 const QUOTA_HYDRATION_FAILURE_TTL: Duration = Duration::from_secs(5);
 const QUOTA_PERIODIC_REFRESH_MIN_INTERVAL: Duration = Duration::from_secs(30 * 60);
-const QUOTA_RESET_RECHECK_GRACE: Duration = Duration::from_secs(30);
-// ready 账号主要依赖真实请求的被动观察；这里只提供低频、定额的状态兜底。
-const QUOTA_READY_REFRESH_MAX_AGE: Duration = Duration::from_secs(2 * 60 * 60);
-const QUOTA_READY_REFRESH_BATCH_SIZE: usize = 8;
 
 /// OpenAI Provider 主动额度刷新的调度策略。
 ///
-/// 账号筛选仍由 core 的可调度状态统一处理；该策略只控制 Provider 自己的
-/// quota worker 频率，避免把 OpenAI 额度语义泄漏到公共调度层。
+/// 正常账号依赖请求响应的被动额度同步；周期 worker 仅复核已耗尽账号。
+/// 该策略只控制 Provider 自己的 quota worker 频率，避免把 OpenAI
+/// 额度语义泄漏到公共调度层。
 #[derive(Debug, Clone, Copy)]
 pub struct CodexQuotaRefreshPolicy {
     interval: Duration,
@@ -197,6 +194,19 @@ pub struct CodexQuotaSyncSummary {
     pub stale: u64,
 }
 
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum QuotaRecoveryEvidence {
+    SuccessfulResponse,
+    RefreshedSnapshot { confirms_recovery: bool },
+    PostFailureSnapshot,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum QuotaRefreshMode {
+    Reconcile,
+    AfterConfirmedFailure,
+}
+
 impl CodexQuotaSyncSummary {
     #[must_use]
     pub const fn has_operational_failures(self) -> bool {
@@ -249,8 +259,6 @@ struct CodexQuotaProjectionState {
     next_version: u64,
     entries: BTreeMap<ProviderAccountId, CodexQuotaSchedulingEntry>,
     last_periodic_refresh_at: BTreeMap<ProviderAccountId, Instant>,
-    last_exhausted_reset_recheck: BTreeMap<ProviderAccountId, CodexQuotaResetMarker>,
-    ready_refresh_cursor: Option<ProviderAccountId>,
 }
 
 #[derive(Debug, Clone, Copy)]
@@ -261,24 +269,6 @@ struct CodexQuotaSchedulingEntry {
     signals: Option<AccountQuotaSignals>,
 }
 
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-struct CodexQuotaResetMarker {
-    revision: CredentialRevision,
-    reset_at: SystemTime,
-}
-
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-enum CodexQuotaRefreshKind {
-    Exhausted { reset_at: Option<SystemTime> },
-    Ready,
-}
-
-#[derive(Clone)]
-struct CodexQuotaRefreshCandidate {
-    account: ProviderAccount,
-    kind: CodexQuotaRefreshKind,
-}
-
 #[derive(Clone)]
 struct CodexQuotaHydrationTarget {
     account: ProviderAccount,
@@ -461,16 +451,15 @@ impl CodexQuotaSchedulingProjection {
     fn reserve_periodic_refreshes(
         &self,
         accounts: Vec<ProviderAccount>,
-        observations: Option<&BTreeMap<ProviderAccountId, QuotaObservation>>,
         now: SystemTime,
     ) -> Vec<ProviderAccount> {
         let candidates = accounts
             .into_iter()
-            .filter_map(|account| quota_refresh_candidate(account, observations, now))
+            .filter_map(|account| quota_refresh_candidate(account, now))
             .collect::<Vec<_>>();
         let candidate_ids = candidates
             .iter()
-            .map(|candidate| candidate.account.id().clone())
+            .map(|account| account.id().clone())
             .collect::<BTreeSet<_>>();
         let refreshed_at = Instant::now();
         let mut state = self
@@ -480,95 +469,27 @@ impl CodexQuotaSchedulingProjection {
         state
             .last_periodic_refresh_at
             .retain(|account_id, _| candidate_ids.contains(account_id));
-        state
-            .last_exhausted_reset_recheck
-            .retain(|account_id, _| candidate_ids.contains(account_id));
 
+        // 正常账号只由真实请求的响应头和 `codex.rate_limits` 被动同步。定时器只兜底
+        // 已耗尽账号，且不依据 reset 时间跳过复核；每个账号仍至少间隔 30 分钟。
         let mut reserved = Vec::new();
-        let mut ready = Vec::new();
-        for candidate in candidates {
-            match candidate.kind {
-                CodexQuotaRefreshKind::Exhausted { reset_at } => {
-                    if reserve_exhausted_quota_refresh(
-                        &mut state,
-                        &candidate.account,
-                        reset_at,
-                        now,
-                        refreshed_at,
-                    ) {
-                        reserved.push(candidate.account);
-                    }
-                }
-                CodexQuotaRefreshKind::Ready => ready.push(candidate.account),
+        for account in candidates {
+            if !periodic_quota_refresh_due(&state, account.id(), refreshed_at) {
+                continue;
             }
-        }
-
-        ready.sort_by(|left, right| left.id().cmp(right.id()));
-        let ready = ready
-            .into_iter()
-            .filter(|account| periodic_quota_refresh_due(&state, account.id(), refreshed_at))
-            .collect::<Vec<_>>();
-        let ready_len = ready.len();
-        if ready_len == 0 {
-            return reserved;
-        }
-        let start = state
-            .ready_refresh_cursor
-            .as_ref()
-            .and_then(|cursor| ready.iter().position(|account| account.id() > cursor))
-            .unwrap_or(0);
-        let ready = ready
-            .into_iter()
-            .cycle()
-            .skip(start)
-            .take(ready_len.min(QUOTA_READY_REFRESH_BATCH_SIZE))
-            .collect::<Vec<_>>();
-        if let Some(last) = ready.last() {
-            state.ready_refresh_cursor = Some(last.id().clone());
-        }
-        for account in &ready {
             state
                 .last_periodic_refresh_at
                 .insert(account.id().clone(), refreshed_at);
+            reserved.push(account);
         }
-        reserved.extend(ready);
         reserved
     }
 }
 
-fn quota_refresh_candidate(
-    account: ProviderAccount,
-    observations: Option<&BTreeMap<ProviderAccountId, QuotaObservation>>,
-    now: SystemTime,
-) -> Option<CodexQuotaRefreshCandidate> {
-    if !eligible_periodic_quota_refresh(&account, now) {
-        return None;
-    }
-    let observation = observations.and_then(|observations| observations.get(account.id()));
-    let snapshot = quota_snapshot_for_account(&account, observation);
-    let kind = match account.availability() {
-        AccountAvailability::QuotaExhausted => CodexQuotaRefreshKind::Exhausted {
-            reset_at: account
-                .cooldown_until()
-                .or_else(|| quota_reset_at_from_snapshot(snapshot.as_ref())),
-        },
-        AccountAvailability::Cooldown
-            if snapshot
-                .as_ref()
-                .is_some_and(|snapshot| snapshot.fact().exhausted()) =>
-        {
-            CodexQuotaRefreshKind::Exhausted {
-                reset_at: quota_reset_at_from_snapshot(snapshot.as_ref()),
-            }
-        }
-        AccountAvailability::Ready
-            if observations.is_some() && quota_observation_is_stale(&account, observation, now) =>
-        {
-            CodexQuotaRefreshKind::Ready
-        }
-        _ => return None,
-    };
-    Some(CodexQuotaRefreshCandidate { account, kind })
+fn quota_refresh_candidate(account: ProviderAccount, now: SystemTime) -> Option<ProviderAccount> {
+    (eligible_periodic_quota_refresh(&account, now)
+        && account.availability() == AccountAvailability::QuotaExhausted)
+        .then_some(account)
 }
 
 fn quota_snapshot_for_account(
@@ -597,72 +518,6 @@ fn quota_observation_confirms_active_exhaustion(
         .is_some_and(|snapshot| quota_fact_is_active_exhaustion(snapshot.fact(), now))
 }
 
-fn quota_observation_is_stale(
-    account: &ProviderAccount,
-    observation: Option<&QuotaObservation>,
-    now: SystemTime,
-) -> bool {
-    let Some(observation) = observation.filter(|observation| {
-        observation.expected_revision == account.revision()
-            && quota_snapshot_from_observation(observation).is_some()
-    }) else {
-        return true;
-    };
-    observation
-        .observed_at
-        .and_then(|observed_at| now.duration_since(observed_at).ok())
-        .is_none_or(|age| age >= QUOTA_READY_REFRESH_MAX_AGE)
-}
-
-fn reserve_exhausted_quota_refresh(
-    state: &mut CodexQuotaProjectionState,
-    account: &ProviderAccount,
-    reset_at: Option<SystemTime>,
-    now: SystemTime,
-    refreshed_at: Instant,
-) -> bool {
-    if let Some(reset_at) = reset_at {
-        if !quota_reset_recheck_ready(reset_at, now) {
-            return false;
-        }
-        let same_reset_already_checked = state
-            .last_exhausted_reset_recheck
-            .get(account.id())
-            .is_some_and(|marker| {
-                marker.revision == account.revision() && marker.reset_at == reset_at
-            });
-        if same_reset_already_checked
-            && !periodic_quota_refresh_due(state, account.id(), refreshed_at)
-        {
-            return false;
-        }
-        state
-            .last_periodic_refresh_at
-            .insert(account.id().clone(), refreshed_at);
-        state.last_exhausted_reset_recheck.insert(
-            account.id().clone(),
-            CodexQuotaResetMarker {
-                revision: account.revision(),
-                reset_at,
-            },
-        );
-        return true;
-    }
-    if !periodic_quota_refresh_due(state, account.id(), refreshed_at) {
-        return false;
-    }
-    state
-        .last_periodic_refresh_at
-        .insert(account.id().clone(), refreshed_at);
-    true
-}
-
-fn quota_reset_recheck_ready(reset_at: SystemTime, now: SystemTime) -> bool {
-    reset_at
-        .checked_add(QUOTA_RESET_RECHECK_GRACE)
-        .is_some_and(|ready_at| ready_at <= now)
-}
-
 fn periodic_quota_refresh_due(
     state: &CodexQuotaProjectionState,
     account_id: &ProviderAccountId,
@@ -774,16 +629,14 @@ impl CodexCredentialQuotaService {
                     .collect::<BTreeMap<_, _>>(),
             ),
             Err(error) => {
-                // 持久观察不可用时不盲扫 ready；保留已耗尽账号的既有恢复轮询。
-                tracing::warn!(error = %error, "OpenAI quota refresh skipped ready-account scan");
+                // 持久观察不可用时跳过旧状态修正；仍复核当前已标记耗尽的账号。
+                tracing::warn!(error = %error, "OpenAI quota refresh skipped persisted exhaustion reconciliation");
                 None
             }
         };
         self.reconcile_persisted_exhaustion(&mut accounts, observations.as_ref(), now)
             .await;
-        let accounts =
-            self.scheduling
-                .reserve_periodic_refreshes(accounts, observations.as_ref(), now);
+        let accounts = self.scheduling.reserve_periodic_refreshes(accounts, now);
         if accounts.is_empty() {
             return Ok(summary);
         }
@@ -796,9 +649,20 @@ impl CodexCredentialQuotaService {
             let observed_at = SystemTime::now();
             match self.fetch_usage_with_recovery(&client, &account).await {
                 Ok(FetchedCodexQuota { account, value }) => {
+                    let previous_snapshot = observations
+                        .as_ref()
+                        .and_then(|observations| observations.get(account.id()))
+                        .filter(|observation| observation.expected_revision == account.revision())
+                        .and_then(quota_snapshot_from_observation);
                     // 单账号解析或落库失败只影响该账号；其余账号继续同步。
                     if let Err(error) = self
-                        .apply_fetched_quota(&account, &value, observed_at, &mut summary)
+                        .apply_fetched_quota(
+                            &account,
+                            &value,
+                            observed_at,
+                            previous_snapshot.as_ref(),
+                            &mut summary,
+                        )
                         .await
                     {
                         summary.transient += 1;
@@ -913,9 +777,16 @@ impl CodexCredentialQuotaService {
         account: &ProviderAccount,
         value: &Value,
         observed_at: SystemTime,
+        previous_snapshot: Option<&CodexAccountQuotaSnapshot>,
         summary: &mut CodexQuotaSyncSummary,
     ) -> Result<(), CodexCredentialQuotaError> {
-        let fact = parse_codex_quota_usage(value)?;
+        let snapshot = parse_account_quota_snapshot(
+            account.id().clone(),
+            account.revision(),
+            observed_at,
+            value,
+        )?;
+        let fact = snapshot.fact();
         let object = value
             .as_object()
             .cloned()
@@ -953,6 +824,9 @@ impl CodexCredentialQuotaService {
             current.cooldown_until(),
             fact,
             SystemTime::now(),
+            QuotaRecoveryEvidence::RefreshedSnapshot {
+                confirms_recovery: quota_snapshot_confirms_recovery(previous_snapshot, &snapshot),
+            },
         ) {
             let _ = self
                 .repository
@@ -1019,6 +893,7 @@ impl CodexCredentialQuotaService {
             current.cooldown_until(),
             fact,
             observed_at,
+            QuotaRecoveryEvidence::SuccessfulResponse,
         ) {
             self.repository
                 .apply_state(
@@ -1094,6 +969,24 @@ impl CodexCredentialQuotaService {
     pub async fn refresh_account(
         &self,
         account_id: &ProviderAccountId,
+    ) -> Result<CodexAccountQuotaSnapshot, CodexCredentialQuotaError> {
+        self.refresh_account_with_mode(account_id, QuotaRefreshMode::Reconcile)
+            .await
+    }
+
+    /// 真实限额失败后的异步刷新只补齐展示快照，不允许同一轮陈旧 usage 撤销失败状态。
+    pub(crate) async fn refresh_account_after_failure(
+        &self,
+        account_id: &ProviderAccountId,
+    ) -> Result<CodexAccountQuotaSnapshot, CodexCredentialQuotaError> {
+        self.refresh_account_with_mode(account_id, QuotaRefreshMode::AfterConfirmedFailure)
+            .await
+    }
+
+    async fn refresh_account_with_mode(
+        &self,
+        account_id: &ProviderAccountId,
+        mode: QuotaRefreshMode,
     ) -> Result<CodexAccountQuotaSnapshot, CodexCredentialQuotaError> {
         let account = self
             .store
@@ -1101,6 +994,7 @@ impl CodexCredentialQuotaService {
             .await?
             .filter(|account| account.provider().as_str() == "openai")
             .ok_or(CodexCredentialQuotaError::NotFound)?;
+        let previous_snapshot = self.read_snapshot_for(&account).await?;
         let observed_at = SystemTime::now();
         if !access_token_is_current(&account, observed_at) {
             return Err(CodexCredentialQuotaError::CredentialRefreshRequired);
@@ -1177,6 +1071,17 @@ impl CodexCredentialQuotaService {
             current.cooldown_until(),
             snapshot.fact(),
             SystemTime::now(),
+            match mode {
+                QuotaRefreshMode::AfterConfirmedFailure => {
+                    QuotaRecoveryEvidence::PostFailureSnapshot
+                }
+                QuotaRefreshMode::Reconcile => QuotaRecoveryEvidence::RefreshedSnapshot {
+                    confirms_recovery: quota_snapshot_confirms_recovery(
+                        previous_snapshot.as_ref(),
+                        &snapshot,
+                    ),
+                },
+            },
         ) {
             self.repository
                 .apply_state(
@@ -1399,9 +1304,13 @@ fn quota_success_state(
     cooldown_until: Option<SystemTime>,
     fact: CodexQuotaFact,
     now: SystemTime,
+    recovery_evidence: QuotaRecoveryEvidence,
 ) -> Option<(AccountAvailability, Option<SystemTime>)> {
     if fact.exhausted() {
-        let observed_cooldown = fact.resets_at().map(SystemTime::from).or(cooldown_until);
+        let observed_cooldown = match (fact.resets_at().map(SystemTime::from), cooldown_until) {
+            (Some(observed), Some(confirmed)) => Some(observed.max(confirmed)),
+            (observed, confirmed) => observed.or(confirmed),
+        };
         return (current != AccountAvailability::QuotaExhausted
             || cooldown_until != observed_cooldown)
             .then_some((AccountAvailability::QuotaExhausted, observed_cooldown));
@@ -1411,8 +1320,13 @@ fn quota_success_state(
         AccountAvailability::Invalid
         | AccountAvailability::Expired
         | AccountAvailability::Banned
-        | AccountAvailability::Unknown
-        | AccountAvailability::QuotaExhausted => Some(AccountAvailability::Ready),
+        | AccountAvailability::Unknown => Some(AccountAvailability::Ready),
+        AccountAvailability::QuotaExhausted
+            if !quota_recovery_confirmed(recovery_evidence, cooldown_until, now) =>
+        {
+            None
+        }
+        AccountAvailability::QuotaExhausted => Some(AccountAvailability::Ready),
         AccountAvailability::Ready => None,
         AccountAvailability::Cooldown
             if cooldown_until.is_some_and(|cooldown_until| cooldown_until <= now) =>
@@ -1424,6 +1338,59 @@ fn quota_success_state(
     availability.map(|availability| (availability, None))
 }
 
+fn quota_recovery_confirmed(
+    evidence: QuotaRecoveryEvidence,
+    cooldown_until: Option<SystemTime>,
+    now: SystemTime,
+) -> bool {
+    if cooldown_until.is_some_and(|cooldown_until| cooldown_until <= now) {
+        return true;
+    }
+    match evidence {
+        QuotaRecoveryEvidence::SuccessfulResponse => true,
+        QuotaRecoveryEvidence::RefreshedSnapshot { confirms_recovery } => confirms_recovery,
+        QuotaRecoveryEvidence::PostFailureSnapshot => false,
+    }
+}
+
+fn quota_snapshot_confirms_recovery(
+    previous: Option<&CodexAccountQuotaSnapshot>,
+    current: &CodexAccountQuotaSnapshot,
+) -> bool {
+    if current.fact().exhausted() {
+        return false;
+    }
+    let Some(previous) = previous else {
+        return false;
+    };
+    if previous
+        .fact()
+        .remaining_percent()
+        .zip(current.fact().remaining_percent())
+        .is_some_and(|(previous, current)| current > previous)
+    {
+        return true;
+    }
+
+    // 单窗口账号可以用同一窗口的用量回落或 reset 前移证明上游提前翻新；多窗口
+    // 无法从账号级 429 确认具体受限窗口，保守等待整体剩余额度改善或真实请求成功。
+    let ([previous], [current]) = (previous.windows(), current.windows()) else {
+        return false;
+    };
+    if previous.key() != current.key() {
+        return false;
+    }
+    let usage_recovered = previous
+        .used_percent()
+        .zip(current.used_percent())
+        .is_some_and(|(previous, current)| current < previous);
+    let window_advanced = previous
+        .reset_at()
+        .zip(current.reset_at())
+        .is_some_and(|(previous, current)| current > previous);
+    usage_recovered || window_advanced
+}
+
 fn quota_state_transition(
     error: &CodexClientError,
     observed_at: SystemTime,

backend/crates/providers/openai/src/provider.rs

@@ -84,6 +84,12 @@ const MAX_COOKIE_HEADER_BYTES: usize = 16 * 1024;
 /// 提交边界前最多保留 64 KiB 原始上游 chunk；达到阈值后结束无感换号窗口，
 /// 但不会把上游数据改写成协议失败。
 const MAX_STREAM_PREFETCH_BYTES: usize = 64 * 1024;
+/// 短暂保留 response.created 等结构事件，让随后到达的明确拒绝可以无感换号；
+/// 到期即放行，避免模型长时间思考时让客户端一直收不到首事件。
+const STREAM_REPLAY_GRACE: Duration = Duration::from_millis(1_200);
+// 额度拒绝后先给上游额度结算留出时间，再以受限时长同步 usage 快照。
+const QUOTA_FAILURE_REFRESH_DELAY: Duration = Duration::from_secs(2);
+const QUOTA_FAILURE_REFRESH_TIMEOUT: Duration = Duration::from_secs(5);
 pub const OFFICIAL_CODEX_BASE_PATH: &str = "/backend-api";
 pub const OFFICIAL_CODEX_BASE_URL: &str = "https://chatgpt.com/backend-api";
 
@@ -679,8 +685,6 @@ fn cold_response_stream(response: ColdResponse) -> EventStream {
             lease.escape_reason(),
             lease.account_switch(),
         );
-        let stream_commit_policy =
-            StreamCommitPolicy::for_continuation(context.continuation_attempt());
         let mut response = create_response_attempt(
             &client,
             &request,
@@ -826,12 +830,13 @@ fn cold_response_stream(response: ColdResponse) -> EventStream {
         let mut decoder = CodexCanonicalDecoder::new(upstream_model.as_str())
             .with_requested_service_tier(request.service_tier())
             .with_raw_sse_passthrough();
-        let mut pre_commit_events = PreCommitClientEvents::new(stream_commit_policy);
+        let mut pre_commit_events = PreCommitClientEvents::new();
         loop {
             let Some(stream_deadline) = remaining(context.deadline()) else {
                 Err(provider_error(ProviderErrorKind::Timeout, UpstreamSendState::Sent))?;
                 return;
             };
+            let replay_grace_deadline = pre_commit_events.replay_grace_deadline();
             let next = tokio::select! {
                 biased;
                 _ = cancellation.cancelled() => Err(MappedProviderFailure::plain(provider_error(
@@ -842,14 +847,21 @@ fn cold_response_stream(response: ColdResponse) -> EventStream {
                     ProviderErrorKind::Timeout,
                     UpstreamSendState::Sent,
                 ))),
+                _ = wait_for_replay_grace(replay_grace_deadline) => Ok(PreCommitPoll::GraceElapsed),
                 chunk = body.next() => match chunk {
-                    Some(Ok(chunk)) => Ok(Some(chunk)),
+                    Some(Ok(chunk)) => Ok(PreCommitPoll::Upstream(Some(chunk))),
                     Some(Err(error)) => Err(map_stream_error(error)),
-                    None => Ok(None),
+                    None => Ok(PreCommitPoll::Upstream(None)),
                 },
             };
             let next = match next {
-                Ok(next) => next,
+                Ok(PreCommitPoll::Upstream(next)) => next,
+                Ok(PreCommitPoll::GraceElapsed) => {
+                    for event in pre_commit_events.commit_pending() {
+                        yield event;
+                    }
+                    continue;
+                }
                 Err(failure) => {
                     let updates = take_rate_limit_updates(rate_limit_updates.as_ref()).await;
                     if !updates.is_empty() {
@@ -1798,54 +1810,37 @@ impl MappedProviderFailure {
     }
 }
 
-/// 与 2.x 一致的流提交策略：普通请求优先首事件，续写/重放优先可恢复性。
-#[derive(Clone, Copy)]
-enum StreamCommitPolicy {
-    FirstForwardableEvent,
-    UntilOutputOrTerminal,
+enum PreCommitPoll<T> {
+    Upstream(T),
+    GraceElapsed,
 }
 
-impl StreamCommitPolicy {
-    const fn for_continuation(attempt: ContinuationAttempt) -> Self {
-        match attempt {
-            ContinuationAttempt::None => Self::FirstForwardableEvent,
-            ContinuationAttempt::Native
-            | ContinuationAttempt::ReplayOwner
-            | ContinuationAttempt::ReplayAny => Self::UntilOutputOrTerminal,
-        }
-    }
-
-    fn boundary_reached(
-        self,
-        events: &[ProviderEvent],
-        timing_signals: ResponseEventSignals,
-        completed: bool,
-    ) -> bool {
-        match self {
-            Self::FirstForwardableEvent => events.iter().any(ProviderEvent::has_client_event),
-            Self::UntilOutputOrTerminal => timing_signals.semantic_output || completed,
-        }
+async fn wait_for_replay_grace(deadline: Option<Instant>) {
+    if let Some(deadline) = deadline {
+        tokio::time::sleep_until(tokio::time::Instant::from_std(deadline)).await;
+    } else {
+        std::future::pending::<()>().await;
     }
 }
 
 /// 提交边界前的上游事件预取。
 ///
-/// 原始 chunk 计数而不是重编码后的 event 大小。64 KiB 只限定无感换号窗口；达到
-/// 阈值会提交已缓存 wire，绝不能因为网关私有资源规则伪造上游协议失败。一旦提交，
-/// 后续事件不再具备无痕重放资格。
+/// 原始 chunk 计数而不是重编码后的 event 大小。时间与 64 KiB 共同限定无感换号
+/// 窗口；任一边界到达都会提交已缓存 wire，不能因网关私有资源规则伪造上游协议
+/// 失败。一旦提交，后续事件不再具备无痕重放资格。
 struct PreCommitClientEvents {
-    policy: StreamCommitPolicy,
     pending: Vec<ProviderEvent>,
     prefetched_bytes: usize,
+    replay_grace_deadline: Option<Instant>,
     committed: bool,
 }
 
 impl PreCommitClientEvents {
-    const fn new(policy: StreamCommitPolicy) -> Self {
+    const fn new() -> Self {
         Self {
-            policy,
             pending: Vec::new(),
             prefetched_bytes: 0,
+            replay_grace_deadline: None,
             committed: false,
         }
     }
@@ -1865,13 +1860,17 @@ impl PreCommitClientEvents {
         if self.committed {
             return incoming;
         }
-        let boundary_reached = self
-            .policy
-            .boundary_reached(&incoming, timing_signals, completed);
+        let starts_replay_grace = incoming.iter().any(ProviderEvent::has_client_event);
         self.pending.extend(incoming);
-        if boundary_reached || self.prefetched_bytes > MAX_STREAM_PREFETCH_BYTES {
+        if timing_signals.semantic_output
+            || completed
+            || self.prefetched_bytes > MAX_STREAM_PREFETCH_BYTES
+        {
             return self.commit_pending();
         }
+        if starts_replay_grace && self.replay_grace_deadline.is_none() {
+            self.replay_grace_deadline = Instant::now().checked_add(STREAM_REPLAY_GRACE);
+        }
         Vec::new()
     }
 
@@ -1900,9 +1899,14 @@ impl PreCommitClientEvents {
     fn take_for_failure(&mut self, incoming: Vec<ProviderEvent>) -> Vec<ProviderEvent> {
         self.pending.extend(incoming);
         self.prefetched_bytes = 0;
+        self.replay_grace_deadline = None;
         std::mem::take(&mut self.pending)
     }
 
+    const fn replay_grace_deadline(&self) -> Option<Instant> {
+        self.replay_grace_deadline
+    }
+
     const fn is_committed(&self) -> bool {
         self.committed
     }
@@ -1910,6 +1914,7 @@ impl PreCommitClientEvents {
     fn commit_pending(&mut self) -> Vec<ProviderEvent> {
         self.committed = true;
         self.prefetched_bytes = 0;
+        self.replay_grace_deadline = None;
         std::mem::take(&mut self.pending)
     }
 }
@@ -1937,7 +1942,7 @@ impl ReplayBoundary {
 struct OpenAiFailureContext<'a> {
     client: &'a CodexBackendClient,
     selector: &'a CodexCredentialSelector,
-    quota: &'a CodexCredentialQuotaService,
+    quota: &'a Arc<CodexCredentialQuotaService>,
     response_origin: &'a Url,
     cyber_policy_scope: Option<&'a CodexCyberPolicyScope>,
     allows_account_state_mutation: bool,
@@ -1952,6 +1957,10 @@ async fn apply_failure(
         return;
     }
     synchronize_passive_quota(context.quota, account, &failure.rate_limit_headers).await;
+    let needs_authoritative_quota_refresh = matches!(
+        failure.account_failure,
+        Some(CodexAccountFailure::QuotaExhausted | CodexAccountFailure::UsageLimitExhausted { .. })
+    );
     if failure.cyber_policy_failure {
         context
             .selector
@@ -1975,6 +1984,9 @@ async fn apply_failure(
             );
         }
     }
+    if needs_authoritative_quota_refresh {
+        schedule_authoritative_quota_refresh_after_failure(context.quota, account);
+    }
     if failure.capture_response_cookies
         && !failure.set_cookie_headers.is_empty()
         && let Err(error) = context
@@ -1994,6 +2006,42 @@ async fn apply_failure(
     }
 }
 
+fn schedule_authoritative_quota_refresh_after_failure(
+    quota: &Arc<CodexCredentialQuotaService>,
+    account: &ProviderAccount,
+) {
+    // 限额错误可以确认账号状态，但 Responses 事件中的 used_percent 可能仍停在上一结算点。
+    // usage 快照在后台补齐展示基线，不得撤销同一轮真实失败，也不能阻塞原始响应。
+    // 先等待 2 秒让上游结算，再查询；5 秒仅限制实际查询自身。
+    let quota = Arc::clone(quota);
+    let account_id = account.id().clone();
+    drop(tokio::spawn(async move {
+        tokio::time::sleep(QUOTA_FAILURE_REFRESH_DELAY).await;
+        match tokio::time::timeout(
+            QUOTA_FAILURE_REFRESH_TIMEOUT,
+            quota.refresh_account_after_failure(&account_id),
+        )
+        .await
+        {
+            Ok(Ok(_)) => {}
+            Ok(Err(error)) => {
+                tracing::warn!(
+                    account_id = %account_id,
+                    error = %error,
+                    "OpenAI authoritative quota synchronization after quota rejection failed"
+                );
+            }
+            Err(_) => {
+                tracing::warn!(
+                    account_id = %account_id,
+                    timeout_ms = QUOTA_FAILURE_REFRESH_TIMEOUT.as_millis(),
+                    "OpenAI authoritative quota synchronization after quota rejection timed out"
+                );
+            }
+        }
+    }));
+}
+
 fn map_handshake_error(error: CodexClientError) -> MappedProviderFailure {
     map_client_error(error, UpstreamSendState::Ambiguous, true)
 }

backend/crates/providers/openai/tests/admin.rs

@@ -20,8 +20,8 @@ use gateway_admin::model::provider_credentials::{
 use gateway_admin::model::{MutationActor, MutationContext, Revision};
 use gateway_admin::ports::provider::ProviderAdminErrorKind;
 use gateway_core::engine::credential::{
-    CredentialRevision, OpaqueProviderData, ProviderAccount, ProviderAccountId,
-    ProviderAccountStore, QuotaObservation,
+    AccountAvailability as CoreAccountAvailability, AccountStateChange, CredentialRevision,
+    OpaqueProviderData, ProviderAccount, ProviderAccountId, ProviderAccountStore, QuotaObservation,
 };
 use gateway_core::operation::{GenerateRequest, Operation, ProtocolPayload};
 use gateway_core::provider_ports::{
@@ -503,6 +503,108 @@ async fn openai_admin_provider_projects_codex_additional_limit_as_the_primary_qu
     assert_eq!(monthly[0].used_percent, Some(2.0));
 }
 
+#[tokio::test]
+async fn openai_admin_projects_confirmed_quota_exhaustion_as_full_without_mutating_raw_usage() {
+    let store = Arc::new(MemoryAccountStore::default());
+    let account_id = "acct_admin_confirmed_exhaustion";
+    store
+        .seed_oauth_credential(ImportCodexOAuthCredential {
+            account_id: account_id.to_owned(),
+            name: "admin confirmed exhaustion".to_owned(),
+            secret: secret("admin-confirmed-exhaustion-access"),
+            verified_account: profile("chatgpt-admin-confirmed-exhaustion"),
+            next_refresh_at: Some(Utc::now() + chrono::Duration::minutes(30)),
+            enabled: true,
+        })
+        .await;
+    let account = store.account(account_id).expect("stored account");
+    let reset_at = 1_900_000_000_u64;
+    store
+        .compare_and_swap_quota(QuotaObservation {
+            account_id: account.id().clone(),
+            expected_revision: account.revision(),
+            quota: Some(OpaqueProviderData::new(
+                json!({
+                    "rate_limit": {
+                        "allowed": true,
+                        "limit_reached": false,
+                        "primary_window": {"used_percent": 86, "reset_at": reset_at}
+                    }
+                })
+                .as_object()
+                .expect("quota object")
+                .clone(),
+            )),
+            observed_at: Some(SystemTime::now()),
+        })
+        .await
+        .expect("persist raw quota");
+    store
+        .apply_state_change(AccountStateChange {
+            account_id: account.id().clone(),
+            expected_revision: account.revision(),
+            availability: CoreAccountAvailability::QuotaExhausted,
+            reason: Some("usage_limit_exhausted".to_owned()),
+            cooldown_until: Some(SystemTime::UNIX_EPOCH + Duration::from_secs(reset_at)),
+            observed_at: SystemTime::now(),
+        })
+        .await
+        .expect("mark account exhausted");
+    let config = valid_config();
+    let bundle = provider_openai::initialize(
+        config.config.clone(),
+        provider_ports_with(Arc::clone(&store), Arc::new(TestOAuthPending::default())),
+    )
+    .await
+    .expect("OpenAI bundle");
+
+    let exhausted = bundle
+        .admin_provider()
+        .quota(ProviderQuotaRequest {
+            account_id: account.id().clone(),
+            refresh: false,
+            rolling_usage: None,
+        })
+        .await
+        .expect("project exhausted quota");
+
+    assert_eq!(exhausted.windows.len(), 1);
+    assert_eq!(exhausted.windows[0].used_percent, Some(100.0));
+    let raw = store
+        .get_quotas(std::slice::from_ref(account.id()))
+        .await
+        .expect("read raw quota")
+        .pop()
+        .expect("raw quota");
+    assert_eq!(
+        raw.quota.expect("raw document").expose_to_provider()["rate_limit"]["primary_window"]["used_percent"],
+        86
+    );
+
+    store
+        .apply_state_change(AccountStateChange {
+            account_id: account.id().clone(),
+            expected_revision: account.revision(),
+            availability: CoreAccountAvailability::Ready,
+            reason: None,
+            cooldown_until: None,
+            observed_at: SystemTime::now(),
+        })
+        .await
+        .expect("recover account");
+    let recovered = bundle
+        .admin_provider()
+        .quota(ProviderQuotaRequest {
+            account_id: account.id().clone(),
+            refresh: false,
+            rolling_usage: None,
+        })
+        .await
+        .expect("project recovered quota");
+
+    assert_eq!(recovered.windows[0].used_percent, Some(86.0));
+}
+
 #[tokio::test]
 async fn openai_admin_provider_rejects_unprepared_mutations_before_store_commit() {
     let store = Arc::new(MemoryAccountStore::default());

backend/crates/providers/openai/tests/config.rs

@@ -117,7 +117,7 @@ fn openai_config_defaults_to_the_provider_owned_operating_values() {
             64,
             8,
             20_000,
-            5,
+            15,
             true,
             true,
             "app_EMoamEEZ73f0CkXaXp7hrann",

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

@@ -498,6 +498,38 @@ async fn manual_quota_refresh_updates_account_state_from_provider_response() {
         AccountAvailability::QuotaExhausted
     );
 
+    server.reset().await;
+    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": 100, "reset_at": 1_900_000_000}
+            }
+        })))
+        .mount(&server)
+        .await;
+
+    let stale = service
+        .refresh_account(account.id())
+        .await
+        .expect("refresh contradictory full quota");
+
+    assert!(!stale.fact().exhausted());
+    assert_eq!(
+        store
+            .account(account_id)
+            .expect("account after contradictory quota")
+            .availability(),
+        AccountAvailability::QuotaExhausted,
+        "an unchanged usage window cannot override a confirmed quota failure"
+    );
+
     server.reset().await;
     Mock::given(method("GET"))
         .and(path("/api/codex/usage"))
@@ -755,7 +787,7 @@ async fn synchronize_without_accounts_is_a_noop_before_network_io() {
 }
 
 #[tokio::test]
-async fn periodic_quota_synchronization_skips_fresh_ready_accounts() {
+async fn periodic_quota_synchronization_does_not_scan_stale_ready_accounts() {
     let store = Arc::new(MemoryAccountStore::default());
     create_account(&store, "acct_periodic_ready").await;
     let account = store
@@ -776,15 +808,16 @@ async fn periodic_quota_synchronization_skips_fresh_ready_accounts() {
                 .expect("quota object")
                 .clone(),
             )),
-            observed_at: Some(SystemTime::now()),
+            // 正常账号即使缓存已旧，也只等待真实请求被动同步，不能被定时任务主动拉取。
+            observed_at: Some(SystemTime::UNIX_EPOCH),
         })
         .await
-        .expect("persist fresh quota");
+        .expect("persist stale quota");
 
     let summary = blocked_network_quota_service(&store)
         .synchronize()
         .await
-        .expect("fresh ready account must not query upstream");
+        .expect("ready account must not query upstream");
 
     assert_eq!(summary, CodexQuotaSyncSummary::default());
 }
@@ -830,7 +863,13 @@ async fn periodic_quota_synchronization_repairs_cooldown_with_active_exhausted_q
     let server = MockServer::start().await;
     Mock::given(method("GET"))
         .and(path("/api/codex/usage"))
-        .respond_with(ResponseTemplate::new(429))
+        .respond_with(ResponseTemplate::new(200).set_body_json(json!({
+            "rate_limit": {
+                "allowed": false,
+                "limit_reached": true,
+                "primary_window": {"used_percent": 100, "reset_at": reset_at}
+            }
+        })))
         .mount(&server)
         .await;
     let service = quota_service_with_base_url(
@@ -939,72 +978,6 @@ async fn periodic_quota_synchronization_skips_ordinary_cooldown_without_exhauste
     );
 }
 
-#[tokio::test]
-async fn periodic_quota_synchronization_limits_stale_ready_account_scans() {
-    let store = Arc::new(MemoryAccountStore::default());
-    for index in 0..9 {
-        let account_id = format!("acct_periodic_stale_ready_{index}");
-        create_account(&store, &account_id).await;
-        let account = store.account(&account_id).expect("created account");
-        store
-            .compare_and_swap_quota(QuotaObservation {
-                account_id: account.id().clone(),
-                expected_revision: account.revision(),
-                quota: Some(OpaqueProviderData::new(
-                    json!({
-                        "rate_limit": {
-                            "allowed": true,
-                            "primary_window": {"used_percent": 10, "reset_at": 1_900_000_000}
-                        }
-                    })
-                    .as_object()
-                    .expect("quota object")
-                    .clone(),
-                )),
-                observed_at: Some(SystemTime::UNIX_EPOCH),
-            })
-            .await
-            .expect("persist stale quota");
-    }
-    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,
-                "primary_window": {"used_percent": 100, "reset_at": 1_900_000_000}
-            }
-        })))
-        .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, 8);
-    assert_eq!(
-        store
-            .account("acct_periodic_stale_ready_0")
-            .expect("scanned account")
-            .availability(),
-        AccountAvailability::QuotaExhausted
-    );
-    assert_eq!(
-        store
-            .account("acct_periodic_stale_ready_8")
-            .expect("deferred account")
-            .availability(),
-        AccountAvailability::Ready
-    );
-}
-
 #[tokio::test]
 async fn periodic_quota_synchronization_attempts_quota_exhausted_accounts() {
     let store = Arc::new(MemoryAccountStore::default());
@@ -1033,7 +1006,7 @@ async fn periodic_quota_synchronization_attempts_quota_exhausted_accounts() {
 }
 
 #[tokio::test]
-async fn periodic_quota_synchronization_waits_for_quota_cooldown_deadline_with_grace() {
+async fn periodic_quota_synchronization_rechecks_exhausted_accounts_without_waiting_for_reset() {
     let store = Arc::new(MemoryAccountStore::default());
     create_account(&store, "acct_periodic_exhausted_cooldown").await;
     let account = store
@@ -1045,38 +1018,29 @@ async fn periodic_quota_synchronization_waits_for_quota_cooldown_deadline_with_g
             expected_revision: account.revision(),
             availability: AccountAvailability::QuotaExhausted,
             reason: Some("usage_limit_exhausted".to_owned()),
-            cooldown_until: SystemTime::now().checked_sub(Duration::from_secs(29)),
+            cooldown_until: SystemTime::now().checked_add(Duration::from_secs(24 * 60 * 60)),
             observed_at: SystemTime::now(),
         })
         .await
-        .expect("mark account exhausted with recent cooldown");
-    let service = blocked_network_quota_service(&store);
+        .expect("mark account exhausted with a future reset");
+    let server = MockServer::start().await;
+    Mock::given(method("GET"))
+        .and(path("/api/codex/usage"))
+        .respond_with(ResponseTemplate::new(500))
+        .expect(1)
+        .mount(&server)
+        .await;
+    let service = quota_service_with_base_url(
+        &store,
+        reqwest::Client::builder()
+            .no_proxy()
+            .build()
+            .expect("client"),
+        server.uri(),
+    );
 
-    let waiting = service
-        .synchronize()
-        .await
-        .expect("quota synchronization before grace");
-
-    assert_eq!(waiting, CodexQuotaSyncSummary::default());
-    let current = store
-        .account("acct_periodic_exhausted_cooldown")
-        .expect("current account");
-    store
-        .apply_state_change(AccountStateChange {
-            account_id: current.id().clone(),
-            expected_revision: current.revision(),
-            availability: AccountAvailability::QuotaExhausted,
-            reason: Some("usage_limit_exhausted".to_owned()),
-            cooldown_until: SystemTime::now().checked_sub(Duration::from_secs(31)),
-            observed_at: SystemTime::now(),
-        })
-        .await
-        .expect("mark account exhausted past grace");
-
-    let due = service
-        .synchronize()
-        .await
-        .expect("quota synchronization after grace");
+    // reset 仅用于展示/调度状态，不能推迟额度耗尽账号的定时复核。
+    let due = service.synchronize().await.expect("quota synchronization");
     let repeated = service
         .synchronize()
         .await
@@ -1086,6 +1050,118 @@ async fn periodic_quota_synchronization_waits_for_quota_cooldown_deadline_with_g
     assert_eq!(repeated, CodexQuotaSyncSummary::default());
 }
 
+#[tokio::test]
+async fn periodic_quota_synchronization_requires_a_real_snapshot_improvement_to_recover() {
+    let store = Arc::new(MemoryAccountStore::default());
+    let account_id = "acct_periodic_recovery_evidence";
+    create_account(&store, account_id).await;
+    let account = store.account(account_id).expect("created account");
+    let reset_at = 1_900_000_000;
+    store
+        .compare_and_swap_quota(QuotaObservation {
+            account_id: account.id().clone(),
+            expected_revision: account.revision(),
+            quota: Some(OpaqueProviderData::new(
+                json!({
+                    "rate_limit": {
+                        "allowed": true,
+                        "limit_reached": false,
+                        "primary_window": {"used_percent": 86, "reset_at": reset_at}
+                    }
+                })
+                .as_object()
+                .expect("quota object")
+                .clone(),
+            )),
+            observed_at: Some(SystemTime::now()),
+        })
+        .await
+        .expect("seed post-failure quota snapshot");
+    store
+        .apply_state_change(AccountStateChange {
+            account_id: account.id().clone(),
+            expected_revision: account.revision(),
+            availability: AccountAvailability::QuotaExhausted,
+            reason: Some("usage_limit_exhausted".to_owned()),
+            cooldown_until: SystemTime::now().checked_add(Duration::from_secs(24 * 60 * 60)),
+            observed_at: SystemTime::now(),
+        })
+        .await
+        .expect("mark account exhausted");
+    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": 86, "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 unchanged = service
+        .synchronize()
+        .await
+        .expect("unchanged quota recheck");
+
+    assert_eq!(unchanged.updated, 1);
+    assert_eq!(
+        store
+            .account(account_id)
+            .expect("account after unchanged snapshot")
+            .availability(),
+        AccountAvailability::QuotaExhausted
+    );
+
+    server.reset().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": 20, "reset_at": reset_at}
+            }
+        })))
+        .expect(1)
+        .mount(&server)
+        .await;
+    // 新 worker 实例模拟下一次 30 分钟复核，避免测试等待真实节流周期。
+    let next_cycle = quota_service_with_base_url(
+        &store,
+        reqwest::Client::builder()
+            .no_proxy()
+            .build()
+            .expect("client"),
+        server.uri(),
+    );
+
+    let improved = next_cycle
+        .synchronize()
+        .await
+        .expect("improved quota recheck");
+
+    assert_eq!(improved.updated, 1);
+    assert_eq!(
+        store
+            .account(account_id)
+            .expect("account after improved snapshot")
+            .availability(),
+        AccountAvailability::Ready
+    );
+}
+
 #[tokio::test]
 async fn periodic_quota_synchronization_throttles_the_same_exhausted_account() {
     let store = Arc::new(MemoryAccountStore::default());
@@ -1113,91 +1189,7 @@ async fn periodic_quota_synchronization_throttles_the_same_exhausted_account() {
 }
 
 #[tokio::test]
-async fn periodic_quota_synchronization_rechecks_an_exhausted_account_once_when_its_reset_is_due() {
-    let store = Arc::new(MemoryAccountStore::default());
-    create_account(&store, "acct_periodic_reset_due").await;
-    let account = store
-        .account("acct_periodic_reset_due")
-        .expect("created account");
-    store
-        .apply_state_change(AccountStateChange {
-            account_id: account.id().clone(),
-            expected_revision: account.revision(),
-            availability: AccountAvailability::QuotaExhausted,
-            reason: Some("quota_exhausted".to_owned()),
-            cooldown_until: None,
-            observed_at: SystemTime::now(),
-        })
-        .await
-        .expect("mark account exhausted");
-    store
-        .compare_and_swap_quota(QuotaObservation {
-            account_id: account.id().clone(),
-            expected_revision: account.revision(),
-            quota: Some(OpaqueProviderData::new(
-                json!({
-                    "rate_limit": {
-                        "allowed": false,
-                        "primary_window": {"used_percent": 100, "reset_at": 1_900_000_000}
-                    }
-                })
-                .as_object()
-                .expect("quota object")
-                .clone(),
-            )),
-            observed_at: Some(SystemTime::now()),
-        })
-        .await
-        .expect("persist future reset");
-    let server = MockServer::start().await;
-    Mock::given(method("GET"))
-        .and(path("/api/codex/usage"))
-        .respond_with(ResponseTemplate::new(500))
-        .mount(&server)
-        .await;
-    let service = quota_service_with_base_url(
-        &store,
-        reqwest::Client::builder()
-            .no_proxy()
-            .build()
-            .expect("client"),
-        server.uri(),
-    );
-
-    let waiting = service.synchronize().await.expect("initial cooldown wait");
-    assert_eq!(waiting, CodexQuotaSyncSummary::default());
-    store
-        .compare_and_swap_quota(QuotaObservation {
-            account_id: account.id().clone(),
-            expected_revision: account.revision(),
-            quota: Some(OpaqueProviderData::new(
-                json!({
-                    "rate_limit": {
-                        "allowed": false,
-                        "primary_window": {"used_percent": 100, "reset_at": 1}
-                    }
-                })
-                .as_object()
-                .expect("quota object")
-                .clone(),
-            )),
-            observed_at: Some(SystemTime::now()),
-        })
-        .await
-        .expect("persist due reset");
-
-    let reset_recheck = service.synchronize().await.expect("reset recheck");
-    let repeated = service
-        .synchronize()
-        .await
-        .expect("throttled reset recheck");
-
-    assert_eq!(reset_recheck.transient, 1);
-    assert_eq!(repeated, CodexQuotaSyncSummary::default());
-}
-
-#[tokio::test]
-async fn periodic_quota_synchronization_reschedules_a_confirmed_new_reset() {
+async fn periodic_quota_synchronization_does_not_bypass_throttle_for_a_new_reset() {
     let store = Arc::new(MemoryAccountStore::default());
     let account_id = "acct_periodic_new_reset";
     create_account(&store, account_id).await;
@@ -1241,7 +1233,10 @@ async fn periodic_quota_synchronization_reschedules_a_confirmed_new_reset() {
         .account(account_id)
         .expect("account after first recheck")
         .cooldown_until();
-    let second = service.synchronize().await.expect("new reset recheck");
+    let second = service
+        .synchronize()
+        .await
+        .expect("throttled reset recheck");
     let repeated = service.synchronize().await.expect("same reset throttled");
     let request_count = server
         .received_requests()
@@ -1260,9 +1255,9 @@ async fn periodic_quota_synchronization_reschedules_a_confirmed_new_reset() {
         (
             1,
             Some(confirmed_reset),
-            1,
+            0,
             CodexQuotaSyncSummary::default(),
-            2,
+            1,
         )
     );
 }

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

@@ -7,7 +7,8 @@ use base64::{Engine as _, engine::general_purpose::STANDARD};
 use chrono::Utc;
 use futures::StreamExt;
 use gateway_core::engine::credential::{
-    AccountAvailability, AccountFeedbackStats, ProviderAccountId, ProviderAccountStore as _,
+    AccountAvailability, AccountFeedbackStats, OpaqueProviderData, ProviderAccountId,
+    ProviderAccountStore as _, QuotaObservation,
 };
 use gateway_core::engine::provider::{Provider as _, ProviderRequest};
 use gateway_core::engine::{
@@ -1006,16 +1007,50 @@ async fn response_failed_before_semantic_output_is_atomic_and_persists_quota_loc
 }
 
 #[tokio::test]
-async fn usage_limit_failure_marks_quota_exhausted_without_usage_probe() {
+async fn usage_limit_failure_returns_promptly_and_refreshes_the_authoritative_quota_snapshot() {
     let store = Arc::new(MemoryAccountStore::default());
     let account_id = "acct_usage_limit_request_path";
     create_account(&store, account_id).await;
+    let account = store.account(account_id).expect("created account");
+    store
+        .compare_and_swap_quota(QuotaObservation {
+            account_id: account.id().clone(),
+            expected_revision: account.revision(),
+            quota: Some(OpaqueProviderData::new(
+                json!({
+                    "rate_limit": {
+                        "allowed": true,
+                        "limit_reached": false,
+                        "primary_window": {"used_percent": 99, "reset_at": 1_900_000_000}
+                    }
+                })
+                .as_object()
+                .expect("stale quota object")
+                .clone(),
+            )),
+            observed_at: Some(SystemTime::now()),
+        })
+        .await
+        .expect("seed stale passive quota");
     let server = MockServer::start().await;
     Mock::given(method("GET"))
         .and(path("/api/codex/usage"))
         .and(header("authorization", format!("Bearer at-{account_id}")))
-        .respond_with(ResponseTemplate::new(500))
-        .expect(0)
+        // 生产环境会短暂返回互相矛盾的快照：数字已到 100%，但布尔位仍声称可用。
+        // 后台同步应覆盖展示数据，不能据此撤销真实请求已经确认的额度耗尽。
+        .respond_with(
+            ResponseTemplate::new(200)
+                // 验证后台 usage 同步不能把原始的额度错误响应拖到查询完成之后。
+                .set_delay(Duration::from_millis(750))
+                .set_body_json(json!({
+                    "rate_limit": {
+                        "allowed": true,
+                        "limit_reached": false,
+                        "primary_window": {"used_percent": 100, "reset_at": 1_900_000_000}
+                    }
+                })),
+        )
+        .expect(1)
         .mount(&server)
         .await;
     Mock::given(method("POST"))
@@ -1040,13 +1075,17 @@ async fn usage_limit_failure_marks_quota_exhausted_without_usage_probe() {
         )
         .await
         .expect("prepare provider stream");
-    let failure = loop {
-        match stream.next().await {
-            Some(Ok(_)) => {}
-            Some(Err(error)) => break error,
-            None => panic!("usage-limit failure must surface a typed failure"),
+    let failure = timeout(Duration::from_millis(300), async {
+        loop {
+            match stream.next().await {
+                Some(Ok(_)) => {}
+                Some(Err(error)) => break error,
+                None => panic!("usage-limit failure must surface a typed failure"),
+            }
         }
-    };
+    })
+    .await
+    .expect("usage refresh must not block the original quota failure");
 
     assert_eq!(failure.kind(), ProviderErrorKind::QuotaExhausted);
     let account = store.account(account_id).expect("usage-limit account");
@@ -1056,22 +1095,69 @@ async fn usage_limit_failure_marks_quota_exhausted_without_usage_probe() {
             .cooldown_until()
             .is_some_and(|until| until > SystemTime::now())
     );
-    assert!(!store.has_quota(account_id));
+    tokio::time::sleep(Duration::from_millis(200)).await;
     let requests = server.received_requests().await.expect("received requests");
     assert_eq!(
         requests
             .iter()
             .filter(|request| request.url.path() == "/api/codex/usage")
             .count(),
-        0
+        0,
+        "usage refresh must wait for the upstream quota settlement delay"
+    );
+
+    let quota = timeout(Duration::from_secs(5), async {
+        loop {
+            let quota = store
+                .get_quotas(&[account.id().clone()])
+                .await
+                .expect("read refreshed quota")
+                .into_iter()
+                .next()
+                .and_then(|observation| observation.quota)
+                .expect("authoritative quota snapshot");
+            let quota = Value::Object(quota.into_inner());
+            if quota
+                .pointer("/rate_limit/primary_window/used_percent")
+                .and_then(Value::as_u64)
+                == Some(100)
+            {
+                break quota;
+            }
+            tokio::time::sleep(Duration::from_millis(10)).await;
+        }
+    })
+    .await
+    .expect("background usage refresh must replace the stale quota snapshot");
+    assert_eq!(
+        quota
+            .pointer("/rate_limit/primary_window/used_percent")
+            .and_then(Value::as_u64),
+        Some(100)
+    );
+    let requests = server.received_requests().await.expect("received requests");
+    assert_eq!(
+        requests
+            .iter()
+            .filter(|request| request.url.path() == "/api/codex/usage")
+            .count(),
+        1
+    );
+    assert_eq!(
+        store
+            .account(account_id)
+            .expect("account after stale full usage refresh")
+            .availability(),
+        AccountAvailability::QuotaExhausted,
+        "a contradictory full usage snapshot must not unlock a confirmed exhausted account"
     );
 }
 
 #[tokio::test]
-async fn ordinary_request_should_forward_created_before_later_failed_chunk() {
+async fn ordinary_request_should_hold_created_until_later_failure_can_rotate() {
     let store = Arc::new(MemoryAccountStore::default());
     create_account(&store, "acct_first_event_latency").await;
-    let (base_url, release, _first_chunk_sent, server) = paused_chunked_sse_server(
+    let (base_url, release, first_chunk_sent, server) = paused_chunked_sse_server(
         concat!(
             "event: response.created\n",
             "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_first_event\",\"model\":\"gpt-5.4\",\"status\":\"in_progress\"}}\n\n"
@@ -1092,41 +1178,111 @@ async fn ordinary_request_should_forward_created_before_later_failed_chunk() {
         .await
         .expect("prepare provider stream");
 
-    let first_event = loop {
-        let next = timeout(Duration::from_secs(1), stream.next())
-            .await
-            .expect("first chunk must not wait for the withheld failure")
-            .expect("provider stream must stay open")
-            .expect("provider event");
-        if next.has_client_event() {
-            break next;
+    let mut first_chunk_sent = Box::pin(first_chunk_sent);
+    loop {
+        tokio::select! {
+            sent = &mut first_chunk_sent => {
+                sent.expect("first upstream chunk");
+                break;
+            }
+            next = stream.next() => {
+                let event = next
+                    .expect("provider stream must stay open")
+                    .expect("provider event");
+                assert!(!event.has_client_event(), "response.created must remain replayable");
+            }
         }
-    };
-    assert_eq!(
-        first_event.wire_event().and_then(|wire| wire.event_type()),
-        Some("response.created")
+    }
+
+    let exposed = timeout(Duration::from_millis(100), async {
+        loop {
+            let next = stream
+                .next()
+                .await
+                .expect("provider stream must stay open")
+                .expect("provider event");
+            if next.has_client_event() {
+                return next;
+            }
+        }
+    })
+    .await;
+    assert!(
+        exposed.is_err(),
+        "a structural event must not commit the downstream before a later 429"
     );
 
     release.send(()).expect("release second upstream chunk");
-    let mut later_wire_types = Vec::new();
-    let failure = loop {
+    let mut failure = loop {
         let next = timeout(Duration::from_secs(1), stream.next())
             .await
             .expect("later failure must arrive after release")
             .expect("provider stream must return typed failure");
         match next {
-            Ok(event) => {
-                if let Some(event_type) = event.wire_event().and_then(|wire| wire.event_type()) {
-                    later_wire_types.push(event_type.to_owned());
-                }
-            }
+            Ok(event) => assert!(
+                !event.has_client_event(),
+                "failure attempt leaked downstream"
+            ),
             Err(error) => break error,
         }
     };
 
-    assert_eq!(later_wire_types, vec!["response.failed"]);
-    assert!(!failure.replay_is_safe());
-    assert!(!failure.has_atomic_client_events());
+    assert!(failure.replay_is_safe());
+    assert_eq!(
+        failure
+            .take_atomic_client_events()
+            .iter()
+            .filter_map(|event| event.wire_event()?.event_type())
+            .collect::<Vec<_>>(),
+        vec!["response.created", "response.failed"]
+    );
+    server.await.expect("chunked SSE server");
+}
+
+#[tokio::test]
+async fn ordinary_request_should_bound_structural_event_replay_grace() {
+    let store = Arc::new(MemoryAccountStore::default());
+    create_account(&store, "acct_bounded_replay_grace").await;
+    let (base_url, release, _first_chunk_sent, server) = paused_chunked_sse_server(
+        concat!(
+            "event: response.created\n",
+            "data: {\"type\":\"response.created\",\"response\":{\"id\":\"resp_bounded_grace\",\"model\":\"gpt-5.4\",\"status\":\"in_progress\"}}\n\n"
+        )
+        .to_owned(),
+        String::new(),
+    )
+    .await;
+    let mut stream = provider_with_base_url(&store, base_url)
+        .execute(
+            planned_request("openai", http_generate_operation()),
+            context("req_bounded_replay_grace", CancellationToken::new()),
+        )
+        .await
+        .expect("prepare provider stream");
+
+    let first_event = timeout(Duration::from_secs(2), async {
+        loop {
+            let event = stream
+                .next()
+                .await
+                .expect("provider stream must stay open")
+                .expect("provider event");
+            if event.has_client_event() {
+                return event;
+            }
+        }
+    })
+    .await
+    .expect("response.created must be released after the bounded grace period");
+
+    assert_eq!(
+        first_event.wire_event().and_then(|wire| wire.event_type()),
+        Some("response.created")
+    );
+    release.send(()).expect("finish upstream response");
+    while let Some(event) = stream.next().await {
+        event.expect("clean upstream EOF");
+    }
     server.await.expect("chunked SSE server");
 }
 

backend/crates/providers/openai/tests/support.rs

@@ -36,6 +36,7 @@ struct StoredAccount {
     account: ProviderAccount,
     credential: gateway_core::engine::credential::PlaintextCredential,
     quota: Option<QuotaObservation>,
+    state_observed_at: Option<SystemTime>,
 }
 
 #[derive(Default)]
@@ -129,6 +130,7 @@ impl ProviderAccountStore for MemoryAccountStore {
                 account: input.account,
                 credential: input.credential,
                 quota: None,
+                state_observed_at: None,
             },
         );
         Ok(())
@@ -260,6 +262,15 @@ impl ProviderAccountStore for MemoryAccountStore {
         if stored.account.revision() != observation.expected_revision {
             return Ok(QuotaWriteOutcome::Conflict);
         }
+        if stored
+            .quota
+            .as_ref()
+            .and_then(|quota| quota.observed_at)
+            .zip(observation.observed_at)
+            .is_some_and(|(current, incoming)| current > incoming)
+        {
+            return Ok(QuotaWriteOutcome::Conflict);
+        }
         stored.quota = Some(observation);
         Ok(QuotaWriteOutcome::Updated)
     }
@@ -272,6 +283,12 @@ impl ProviderAccountStore for MemoryAccountStore {
         if stored.account.revision() != change.expected_revision {
             return Err(store_error(StoreErrorKind::Conflict));
         }
+        if stored
+            .state_observed_at
+            .is_some_and(|current| current > change.observed_at)
+        {
+            return Err(store_error(StoreErrorKind::Conflict));
+        }
         stored.account = rebuild_account(
             &stored.account,
             AccountRebuild {
@@ -285,6 +302,7 @@ impl ProviderAccountStore for MemoryAccountStore {
                 profile: None,
             },
         );
+        stored.state_observed_at = Some(change.observed_at);
         Ok(())
     }
 

backend/crates/providers/xai/src/credential/catalog.rs

@@ -469,7 +469,7 @@ impl GrokCredentialQuotaService {
         if quota_refresh_may_update_state(&loaded.account, &current.account)
             && let Some(availability) = quota_refresh_availability(
                 current.account.availability(),
-                quota_is_exhausted(snapshot.billing()),
+                authoritative_quota_exhaustion(snapshot.billing()),
             )
         {
             self.repository
@@ -1329,6 +1329,13 @@ fn quota_is_exhausted(billing: &GrokBillingPresentation) -> bool {
         )
 }
 
+/// `None` means the billing document cannot prove either quota state.
+fn authoritative_quota_exhaustion(billing: &GrokBillingPresentation) -> Option<bool> {
+    billing
+        .has_authoritative_quota()
+        .then(|| quota_is_exhausted(billing))
+}
+
 fn quota_refresh_may_update_state(before: &ProviderAccount, current: &ProviderAccount) -> bool {
     current.enabled()
         && current
@@ -1341,22 +1348,26 @@ fn quota_refresh_may_update_state(before: &ProviderAccount, current: &ProviderAc
 
 fn quota_refresh_availability(
     current: AccountAvailability,
-    exhausted: bool,
+    exhausted: Option<bool>,
 ) -> Option<GrokCredentialAvailability> {
     match current {
         AccountAvailability::Invalid
         | AccountAvailability::Expired
         | AccountAvailability::Banned => None,
         AccountAvailability::QuotaExhausted => {
-            (!exhausted).then_some(GrokCredentialAvailability::Ready)
+            matches!(exhausted, Some(false)).then_some(GrokCredentialAvailability::Ready)
+        }
+        AccountAvailability::Cooldown | AccountAvailability::Unknown => {
+            exhausted.map(|exhausted| {
+                if exhausted {
+                    GrokCredentialAvailability::QuotaExhausted
+                } else {
+                    GrokCredentialAvailability::Ready
+                }
+            })
         }
-        AccountAvailability::Cooldown | AccountAvailability::Unknown => Some(if exhausted {
-            GrokCredentialAvailability::QuotaExhausted
-        } else {
-            GrokCredentialAvailability::Ready
-        }),
         AccountAvailability::Ready => {
-            exhausted.then_some(GrokCredentialAvailability::QuotaExhausted)
+            matches!(exhausted, Some(true)).then_some(GrokCredentialAvailability::QuotaExhausted)
         }
     }
 }

backend/crates/providers/xai/tests/credential/catalog.rs

@@ -874,17 +874,24 @@ async fn quota_projection_preserves_unknown_period_for_dynamic_duration_fallback
 }
 
 #[tokio::test]
-async fn weekly_period_without_reported_allowance_is_not_authoritative_quota() {
-    let (_, repository) = repository_with_accounts(&[("free-quota", "subject-free-quota")]).await;
+async fn non_authoritative_quota_refresh_does_not_clear_existing_quota_exhaustion() {
+    let (store, repository) =
+        repository_with_accounts(&[("free-quota", "subject-free-quota")]).await;
+    let id = account_id("free-quota");
+    set_account_state(&store, &id, AccountAvailability::QuotaExhausted, None).await;
     let transport = QueueBillingTransport::success(
         br#"{"config":{"currentPeriod":{"type":"USAGE_PERIOD_TYPE_WEEKLY","start":"2026-07-15T00:00:00Z","end":"2026-07-22T00:00:00Z"},"onDemandCap":{"val":0},"onDemandUsed":{"val":0},"prepaidBalance":{"val":0}}}"#,
     );
     let snapshot = crate::support::grok_quota_service(repository, transport)
-        .refresh_account(&account_id("free-quota"))
+        .refresh_account(&id)
         .await
         .expect("refresh Free quota");
 
     assert!(!snapshot.billing().has_authoritative_quota());
+    assert_eq!(
+        store.account(&id).expect("account").availability(),
+        AccountAvailability::QuotaExhausted
+    );
 }
 
 #[tokio::test]

backend/migrations/.frozen-sha256

@@ -1,5 +1 @@
-0554fed953bd1440b50bc4033dc8bad96b51e132c57c2ccf1553728a991d5cb2  0001_initial.sql
-0ae3945c02461347ab8dcfc64821027cd2366505778e0450d71bd82772c4e99d  0002_drop_opaque_response_id_index.sql
-e608ed89991f9d6317491714447a00a61ede90caca4fb030eb30695654afb664  0003_store_opaque_response_ids_as_bytes.sql
-df4e02cf2335733326c12429ced4cbd0b35b8a22b3fc24981c3c0549477d7c43  0004_add_model_request_service_tier.sql
-fe8f36913c6db7053148db450f05e008dab6cdb9ca8cdcfcb22d97ad9a87bf49  0005_allow_quota_exhausted_cooldown.sql
+a4681a5c040e80c5234e879293fc34972080e5251dfbc7c40e1374f8e4170a4c  0001_initial.sql

backend/migrations/0001_initial.sql

@@ -166,7 +166,14 @@ create table provider_accounts (
     )
   ),
   constraint provider_accounts_cooldown_ck check (
-    (availability = 'cooldown') = (cooldown_until is not null)
+    (
+      availability != 'cooldown'
+      or cooldown_until is not null
+    )
+    and (
+      cooldown_until is null
+      or availability in ('cooldown', 'quota_exhausted')
+    )
   ),
   constraint provider_accounts_time_ck check (
     created_at <= updated_at
@@ -208,6 +215,7 @@ create table model_requests (
   requested_model_id text not null,
   provider_kind text,
   upstream_model_id text,
+  service_tier text,
   provider_account_id text,
   provider_account_ref text,
   upstream_transport text,
@@ -220,9 +228,11 @@ create table model_requests (
   outcome text not null default 'running',
   client_status_code integer,
   upstream_status_code integer,
-  client_response_id text,
+  -- Provider response IDs are opaque JSON strings. Bytes preserve embedded NUL
+  -- and avoid imposing PostgreSQL text-index limits on protocol handles.
+  client_response_id bytea,
   upstream_request_id text,
-  upstream_response_id text,
+  upstream_response_id bytea,
   error_kind text,
   provider_error_code text,
   error_message text,
@@ -321,6 +331,13 @@ create table model_requests (
     provider_observation_json is null
     or jsonb_typeof(provider_observation_json) = 'object'
   ),
+  constraint model_requests_service_tier_ck check (
+    service_tier is null
+    or (
+      octet_length(service_tier) between 1 and 64
+      and service_tier !~ '[[:cntrl:]]'
+    )
+  ),
   constraint model_requests_cost_ck check (
     cost_source in ('provider_reported', 'calculated', 'unavailable')
     and (
@@ -414,10 +431,6 @@ create index model_requests_running_deadline_idx
 create index model_requests_retention_idx
   on model_requests (completed_at)
   where outcome <> 'running';
-create unique index model_requests_client_response_uq
-  on model_requests (client_response_id)
-  where client_response_id is not null;
-
 create table ops_events (
   id text primary key,
   model_request_id text,

backend/migrations/0002_drop_opaque_response_id_index.sql

@@ -1,3 +0,0 @@
--- Provider response IDs are opaque protocol values, not relational identifiers.
--- Indexing the raw text makes PostgreSQL reject otherwise valid long handles.
-drop index model_requests_client_response_uq;

backend/migrations/0003_store_opaque_response_ids_as_bytes.sql

@@ -1,7 +0,0 @@
--- Response IDs are opaque JSON strings. PostgreSQL text rejects NUL even though JSON accepts it.
--- Store their UTF-8 bytes without an index; Rust decodes them only at the audit/API boundary.
-alter table model_requests
-  alter column client_response_id type bytea
-    using convert_to(client_response_id, 'UTF8'),
-  alter column upstream_response_id type bytea
-    using convert_to(upstream_response_id, 'UTF8');

backend/migrations/0004_add_model_request_service_tier.sql

@@ -1,9 +0,0 @@
-alter table model_requests
-  add column service_tier text,
-  add constraint model_requests_service_tier_ck check (
-    service_tier is null
-    or (
-      octet_length(service_tier) between 1 and 64
-      and service_tier !~ '[[:cntrl:]]'
-    )
-  );

backend/migrations/0005_allow_quota_exhausted_cooldown.sql

@@ -1,12 +0,0 @@
-alter table provider_accounts
-  drop constraint provider_accounts_cooldown_ck,
-  add constraint provider_accounts_cooldown_ck check (
-    (
-      availability != 'cooldown'
-      or cooldown_until is not null
-    )
-    and (
-      cooldown_until is null
-      or availability in ('cooldown', 'quota_exhausted')
-    )
-  );

backend/migrations/README.md

@@ -6,14 +6,16 @@ sqlx 在服务启动时（`serve` 监听之前）按编号顺序执行本目录
 
 ## 冻结规则
 
-- 已合入 main 的迁移文件**永久冻结**，一个字节都不能改——包括注释和空白。
-  任何改动都会让所有已部署实例（含自更新链路上的存量用户）在下次重启时
-  因 checksum 失配而无法启动，且没有自动恢复路径。
-- schema 变更一律新增编号迁移（`0003_...`、`0004_...`），永远不回改旧文件。
+- 同一大版本内，已合入 main 的迁移文件**永久冻结**，一个字节都不能改——包括注释和
+  空白。任何改动都会让该大版本的已部署实例在下次重启时因 checksum 失配而无法启动，
+  且没有自动恢复路径。
+- 同一大版本内的 schema 变更一律新增编号迁移（`0002_...`、`0003_...`），永远不回改
+  旧文件。切换大版本时允许把已完成升级的 schema 折叠成新的 `0001` 基线，因为跨大版本
+  不支持在线升级，目标版本必须使用全新数据目录部署。
 - `.frozen-sha256` 是冻结清单（`sha256sum` 格式）。CI 会校验：清单内文件
   字节未变，且目录里每个 `*.sql` 都已入册。新增迁移时在同一提交里执行
   `sha256sum 000N_xxx.sql >> .frozen-sha256` 入册。
-- 若确需修正已冻结迁移里的错误，用新迁移做补偿性变更（`alter` / 回填），
+- 若同一大版本内确需修正已冻结迁移里的错误，用新迁移做补偿性变更（`alter` / 回填），
   不要动原文件。
 
 ## 本地测试库

deploy/README.md

@@ -130,10 +130,8 @@ Provider schema 以明文 JSON 保存于 PostgreSQL。Redis 只保存可重建
 ## 镜像升级与源码构建
 
 > [!WARNING]
-> 以下命令只适用于同一大版本内的升级。数据库 schema 与配置只在同一大版本内兼容，
-> 跨大版本直接换镜像会导致服务无法启动；内置一键自更新会拒绝跨大版本目标（3.0 之前
-> 的旧版没有这道门禁，不要用旧版界面的一键更新跨大版本）。跨大版本请全新部署（全新
-> `.runtime/` 数据目录），并重新导入或重新授权 Provider 账号与客户端 Key。
+> 以下命令只适用于同一大版本内的升级，不支持跨大版本在线升级。跨大版本请使用全新的
+> `.runtime/` 数据目录重新部署，并重新导入或重新授权 Provider 账号与客户端 Key。
 
 ```bash
 docker compose -f deploy/compose.yaml build codex-proxy-rs

docs/architecture.md

@@ -170,11 +170,11 @@ Admin API 不要求客户端提交配置 revision，也不向客户端暴露配
   客户端可见协议结果。队列不是第二份业务权威，也不做无幂等键的盲目重试。
 - retention 只删除已满足保留规则的历史事实，不改变运行中请求。
 
-业务 schema 由 `backend/migrations/` 下的编号迁移按序定义：`0001_initial.sql` 建表，`0002` 删除
-opaque response ID 唯一索引，`0003` 将 client/upstream response ID 改为无格式假设的 UTF-8 bytes，
-`0004` 增加响应观测得到的 `model_requests.service_tier`。已合入的迁移永久冻结——
-`.frozen-sha256` 是冻结清单，CI 校验清单相对 PR base 只增不改，启动时 sqlx 重校验已应用迁移的
-checksum，不一致直接拒绝启动。规则见 `backend/migrations/README.md`。
+业务 schema 由 `backend/migrations/0001_initial.sql` 直接定义当前大版本的完整新库基线，其中
+response ID 使用无格式假设的 UTF-8 bytes，`model_requests.service_tier` 保存响应观测到的服务档位，
+且额度耗尽账号可以携带冷却截止时间。同一大版本内已合入的迁移永久冻结——
+`.frozen-sha256` 是冻结清单，CI 校验清单相对 PR base 只增不改，启动时 sqlx 重校验已应用
+迁移的 checksum，不一致直接拒绝启动。规则见 `backend/migrations/README.md`。
 
 ## 9. Redis 终态
 

frontend/src/views/accounts/components/AccountOverviewCards.vue

@@ -33,7 +33,7 @@ const overviewItems = computed(() => [
     icon: Gauge,
   },
   {
-    label: '不可用账号',
+    label: '异常账号',
     value: formatCount(props.summary.unavailable),
     caption: '过期 / 禁用 / 封禁',
     tone: 'danger',

release/version.yaml

@@ -1 +1 @@
-version: "3.0.28"
+version: "3.0.30"

skills/codex-proxy-rs/references/repo-guide.md

@@ -40,9 +40,9 @@
 
 ## 存储
 
-- PostgreSQL 业务表只有 `0001_initial.sql` 定义的七张；后续编号迁移只做增量。当前 `0002` 删除
-  opaque response ID 索引，`0003` 把 response ID 改为 bytes，`0004` 增加响应观测的 service tier。
-- 已应用迁移按字节冻结：`backend/migrations/.frozen-sha256` 是冻结清单，CI 相对 PR base 做 append-only 校验；schema 变更一律新增编号迁移，规则见 `backend/migrations/README.md`。
+- PostgreSQL 业务表只有 `0001_initial.sql` 定义的七张；该文件直接表达当前大版本的完整新库基线，
+  包括 bytes response ID、响应观测 service tier 与额度耗尽冷却约束。
+- 同一大版本内已应用迁移按字节冻结：`backend/migrations/.frozen-sha256` 是冻结清单，CI 相对 PR base 做 append-only 校验；schema 变更一律新增编号迁移，规则见 `backend/migrations/README.md`。
 - `config_revision` 只用于会改变调度快照或安全配置的管理 mutation。
 - quota、cooldown、catalog generation、自动 refresh 不推进全局 revision。
 - refresh 只推进账号 `credential_revision`；Redis cooldown 是可丢失热缓存。