fix(openai): verify idle quota windows before warmup (#653)

Co-authored-by: Zyy <zyycn@users.noreply.github.com>
Co-authored-by: Codex <noreply@openai.com>

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

@@ -52,6 +52,7 @@ pub struct CodexQuotaWindow {
     window_seconds: Option<u64>,
     used_percent: Option<f64>,
     reset_at: Option<DateTime<Utc>>,
+    reset_after_seconds: Option<u64>,
     limit_reached: bool,
 }
 
@@ -102,6 +103,11 @@ impl CodexQuotaWindow {
         self.reset_at
     }
 
+    /// 上游观察时的剩余秒数，不能用当前时间重新推算空闲窗口
+    pub(super) const fn reset_after_seconds(&self) -> Option<u64> {
+        self.reset_after_seconds
+    }
+
     /// 该窗口是否已触顶（`limit_reached` 或 `used_percent >= 100`）
     #[must_use]
     pub fn limit_reached(&self) -> bool {
@@ -573,6 +579,14 @@ fn parse_rate_limit_windows(
             })
             .transpose()?;
         let limit_reached = optional_bool(window, "limit_reached")?.unwrap_or(false);
+        let reset_after_seconds = window
+            .get("reset_after_seconds")
+            .map(|value| {
+                value
+                    .as_u64()
+                    .ok_or(CodexCredentialQuotaError::InvalidCredentialData)
+            })
+            .transpose()?;
         let kind = quota_window_kind(window_seconds);
         // primary/secondary 是协议槽位，不代表固定周期
         // 窗口身份使用上游
@@ -591,6 +605,7 @@ fn parse_rate_limit_windows(
             window_seconds,
             used_percent,
             reset_at,
+            reset_after_seconds,
             limit_reached,
         });
     }

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

@@ -2,6 +2,86 @@
 
 use super::*;
 
+// 上游倒计时按整秒返回，跨过一个完整秒后再区分刚激活与空闲满窗口
+const WINDOW_CONFIRM_DELAY: Duration = Duration::from_secs(2);
+
+enum WarmupDecision<'a> {
+    Ready,
+    Active(&'a CodexQuotaWindow),
+    Exhausted(Option<&'a CodexQuotaWindow>),
+    Verify(&'a CodexQuotaWindow),
+}
+
+fn warmup_account_is_available(account: &ProviderAccount) -> bool {
+    account.authentication_kind() == crate::credential::CODEX_AUTHENTICATION_KIND_OAUTH
+        && account.enabled()
+        && matches!(
+            account.credential_state(),
+            CredentialState::Unknown | CredentialState::Ready
+        )
+}
+
+fn warmup_decision<'a>(
+    snapshot: &'a CodexAccountQuotaSnapshot,
+    previous: Option<&CodexAccountQuotaSnapshot>,
+    now: chrono::DateTime<chrono::Utc>,
+) -> WarmupDecision<'a> {
+    if snapshot.quota().is_exhausted() {
+        return WarmupDecision::Exhausted(None);
+    }
+    if let Some(window) = snapshot.windows().iter().find(|window| {
+        window.is_account_wide()
+            && window.kind() == CodexQuotaWindowKind::Weekly
+            && window.limit_reached()
+    }) {
+        return WarmupDecision::Exhausted(Some(window));
+    }
+
+    let mut uncertain = None;
+    for window in snapshot.windows().iter().filter(|window| {
+        window.is_account_wide()
+            && window.kind() == CodexQuotaWindowKind::ShortTerm
+            && window
+                .reset_at()
+                .is_some_and(|reset| reset > now + chrono::Duration::minutes(30))
+    }) {
+        // 0% 可能只是取整；已经开始递减的原始倒计时仍能证明窗口已激活
+        if window.used_percent().is_some_and(|used| used > 0.0)
+            || window
+                .reset_after_seconds()
+                .zip(window.window_seconds())
+                .is_some_and(|(remaining, total)| remaining > 0 && remaining < total)
+        {
+            return WarmupDecision::Active(window);
+        }
+        // 连续两次实时查询仍返回完整窗口且 reset 向后移动，才确认空闲
+        // 不以快照级 observed_at 推算，其他额度桶的被动更新也会推进该时间
+        let confirmed_idle = idle_full_window(window)
+            && previous.is_some_and(|previous| {
+                previous.windows().iter().any(|prior| {
+                    prior.key() == window.key()
+                        && idle_full_window(prior)
+                        && prior
+                            .reset_at()
+                            .zip(window.reset_at())
+                            .is_some_and(|(before, after)| after > before)
+                })
+            });
+        if !confirmed_idle {
+            uncertain = Some(window);
+        }
+    }
+    uncertain.map_or(WarmupDecision::Ready, WarmupDecision::Verify)
+}
+
+fn idle_full_window(window: &CodexQuotaWindow) -> bool {
+    window.used_percent() == Some(0.0)
+        && window
+            .reset_after_seconds()
+            .zip(window.window_seconds())
+            .is_some_and(|(remaining, total)| remaining == total)
+}
+
 #[derive(Clone, Copy, PartialEq, Eq)]
 enum WarmupTerminal {
     Missing,
@@ -77,14 +157,7 @@ impl CodexCredentialQuotaService {
         model: &str,
     ) -> Result<CodexWarmupSummary, CodexCredentialQuotaError> {
         let mut accounts = self.repository.list_for_provider().await?;
-        accounts.retain(|account| {
-            account.authentication_kind() == crate::credential::CODEX_AUTHENTICATION_KIND_OAUTH
-                && account.enabled()
-                && matches!(
-                    account.credential_state(),
-                    CredentialState::Unknown | CredentialState::Ready
-                )
-        });
+        accounts.retain(warmup_account_is_available);
         let mut summary = CodexWarmupSummary::default();
         if accounts.is_empty() {
             return Ok(summary);
@@ -94,7 +167,7 @@ impl CodexCredentialQuotaService {
             .map(|account| account.id().clone())
             .collect::<Vec<_>>();
         let observed = self.store.get_quotas(&account_ids).await?;
-        let observed_snapshots = observed
+        let mut observed_snapshots = observed
             .iter()
             .filter_map(|obs| {
                 quota_snapshot_from_observation(obs)
@@ -108,35 +181,108 @@ impl CodexCredentialQuotaService {
             self.profile.clone(),
         );
 
-        let now_utc = chrono::Utc::now();
-        for account in accounts {
+        'accounts: for account in accounts {
             if account.quota().is_exhausted() {
                 summary.skipped_exhausted += 1;
+                tracing::info!(
+                    account_id = %account.id(),
+                    observed_at = ?account.quota().observed_at(),
+                    reset_at = ?account.quota().reset_at(),
+                    reason = "quota_exhausted",
+                    "OpenAI account warmup skipped"
+                );
                 continue;
             }
-            if let Some(snapshot) = observed_snapshots.get(account.id()) {
-                // 1. 周线已触顶或耗尽跳过
-                let weekly_exhausted = snapshot
-                    .windows()
-                    .iter()
-                    .any(|w| w.kind() == CodexQuotaWindowKind::Weekly && w.limit_reached());
-                if weekly_exhausted {
-                    summary.skipped_exhausted += 1;
-                    continue;
-                }
-                // 2. 5h 窗口当前活跃且距重置时间 > 30 分钟跳过
-                let has_active_5h = snapshot.windows().iter().any(|w| {
-                    w.kind() == CodexQuotaWindowKind::ShortTerm
-                        && w.reset_at().is_some_and(|reset_at| {
-                            reset_at > now_utc + chrono::Duration::minutes(30)
-                        })
-                });
-                if has_active_5h {
-                    summary.skipped_active += 1;
-                    continue;
-                }
+            let mut snapshot = observed_snapshots
+                .remove(account.id())
+                .filter(|snapshot| snapshot.credential_revision() == account.revision());
+            let mut previous = None;
+            let mut refreshes = 0;
+            while let Some(current) = snapshot.as_ref() {
+                let (window, reason) =
+                    match warmup_decision(current, previous.as_ref(), chrono::Utc::now()) {
+                        WarmupDecision::Ready => break,
+                        WarmupDecision::Active(window) => {
+                            summary.skipped_active += 1;
+                            (Some(window), "active_short_window")
+                        }
+                        WarmupDecision::Exhausted(window) => {
+                            summary.skipped_exhausted += 1;
+                            (window, "quota_exhausted")
+                        }
+                        WarmupDecision::Verify(window) if refreshes == 2 => {
+                            summary.failed += 1;
+                            (Some(window), "window_activity_unconfirmed")
+                        }
+                        WarmupDecision::Verify(_) => {
+                            // 持久化快照不作为两次实时查询的首份证据
+                            if refreshes == 1 {
+                                previous = snapshot.take();
+                                tokio::time::sleep(WINDOW_CONFIRM_DELAY).await;
+                            }
+                            match tokio::time::timeout(
+                                WARMUP_REQUEST_TIMEOUT,
+                                self.refresh_account(account.id()),
+                            )
+                            .await
+                            {
+                                Ok(Ok(refreshed))
+                                    if refreshed.credential_revision() == account.revision() =>
+                                {
+                                    snapshot = Some(refreshed);
+                                    refreshes += 1;
+                                    continue;
+                                }
+                                _ => {
+                                    summary.failed += 1;
+                                    tracing::warn!(
+                                        account_id = %account.id(),
+                                        reason = "quota_refresh_failed",
+                                        "OpenAI account warmup quota verification failed"
+                                    );
+                                    continue 'accounts;
+                                }
+                            }
+                        }
+                    };
+                tracing::info!(
+                    account_id = %account.id(),
+                    observed_at = ?chrono::DateTime::<chrono::Utc>::from(current.observed_at()),
+                    window = window.map(CodexQuotaWindow::key),
+                    window_seconds = window.and_then(CodexQuotaWindow::window_seconds),
+                    used_percent = window.and_then(CodexQuotaWindow::used_percent),
+                    reset_at = ?window.and_then(CodexQuotaWindow::reset_at),
+                    reset_after_seconds = window.and_then(CodexQuotaWindow::reset_after_seconds),
+                    reason,
+                    "OpenAI account warmup skipped"
+                );
+                continue 'accounts;
             }
 
+            // 回查会更新额度与套餐；凭据加载仍核对最新账号的完整快照
+            let account = if refreshes > 0 {
+                match self.store.get_account(account.id()).await {
+                    Ok(Some(current))
+                        if current.provider() == account.provider()
+                            && current.revision() == account.revision()
+                            && warmup_account_is_available(&current)
+                            && !current.quota().is_exhausted() =>
+                    {
+                        current
+                    }
+                    _ => {
+                        summary.failed += 1;
+                        tracing::warn!(
+                            account_id = %account.id(),
+                            reason = "account_changed_after_verification",
+                            "OpenAI account warmup skipped"
+                        );
+                        continue;
+                    }
+                }
+            } else {
+                account
+            };
             let credential = match self.repository.load_runtime_credential(&account).await {
                 Ok(cred) => cred,
                 Err(_) => {

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

@@ -1,6 +1,314 @@
-//! 验证额度预热按 SSE 终态判定结果、请求档位并跳过过期凭据
+//! 验证额度预热的窗口识别、实时复核、SSE 终态与凭据边界
 
 use super::*;
+use gateway_core::account::{OpaqueProviderData, QuotaObservation, QuotaWriteOutcome};
+
+fn short_window(used: u64, remaining: Option<u64>, reset: i64) -> serde_json::Value {
+    let mut window = json!({
+        "used_percent": used,
+        "limit_window_seconds": 18_000,
+        "reset_at": reset,
+    });
+    if let Some(remaining) = remaining {
+        window["reset_after_seconds"] = json!(remaining);
+    }
+    window
+}
+
+fn usage(window: serde_json::Value) -> serde_json::Value {
+    json!({"rate_limit": {"allowed": true, "primary_window": window}})
+}
+
+async fn seed_quota(
+    store: &MemoryAccountStore,
+    account_id: &str,
+    quota: serde_json::Value,
+    observed_at: SystemTime,
+) {
+    let account = store.account(account_id).expect("account");
+    assert_eq!(
+        store
+            .compare_and_swap_quota(QuotaObservation {
+                expected_connection: None,
+                plan_type: None,
+                account_id: account.id().clone(),
+                expected_revision: account.revision(),
+                quota: OpaqueProviderData::new(quota.as_object().unwrap().clone()),
+                observed_at,
+                state: QuotaState::allowed(observed_at),
+            })
+            .await
+            .expect("persist quota"),
+        QuotaWriteOutcome::Updated
+    );
+}
+
+async fn mount_completion(server: &MockServer, expected: u64) {
+    Mock::given(method("POST"))
+        .and(path("/codex/responses"))
+        .respond_with(ResponseTemplate::new(200).set_body_raw(
+            "event: response.completed\ndata: {\"type\":\"response.completed\",\"response\":{\"status\":\"completed\"}}\n\n",
+            "text/event-stream",
+        ))
+        .expect(expected)
+        .mount(server)
+        .await;
+}
+
+#[tokio::test]
+async fn warmup_uses_account_windows_and_preserves_the_reset_threshold() {
+    for (name, used, remaining, until_reset, named, warmed) in [
+        ("acct_zero_active", 0, Some(17_999), 17_999, false, false),
+        ("acct_passive_active", 1, None, 14_400, false, false),
+        ("acct_near_reset", 1, Some(1_800), 1_800, false, true),
+        ("acct_expired", 1, Some(3_600), -1, false, true),
+        ("acct_reset_zero", 0, Some(0), -1, false, true),
+        ("acct_named_active", 1, Some(3_600), 3_600, true, true),
+    ] {
+        let server = MockServer::start().await;
+        let store = Arc::new(MemoryAccountStore::default());
+        create_account(&store, name).await;
+        let window = short_window(used, remaining, Utc::now().timestamp() + until_reset);
+        let quota = if named {
+            json!({
+                "rate_limit": {"allowed": true},
+                "additional_rate_limits": [{
+                    "limit_id": "other_feature",
+                    "rate_limit": {
+                        "primary_window": window,
+                        "secondary_window": {
+                            "used_percent": 100,
+                            "limit_window_seconds": 604_800,
+                            "reset_at": Utc::now().timestamp() + 604_800,
+                        },
+                    },
+                }],
+            })
+        } else {
+            usage(window)
+        };
+        seed_quota(&store, name, quota, SystemTime::now()).await;
+        mount_completion(&server, u64::from(warmed)).await;
+        Mock::given(method("GET"))
+            .respond_with(ResponseTemplate::new(500))
+            .expect(0)
+            .mount(&server)
+            .await;
+        let service = quota_service_with_base_url(&store, reqwest::Client::new(), server.uri());
+
+        let summary = service.execute_warmup("test-model").await.unwrap();
+
+        assert_eq!(
+            (
+                summary.warmed_up,
+                summary.skipped_active,
+                summary.skipped_exhausted,
+                summary.failed
+            ),
+            (u64::from(warmed), u64::from(!warmed), 0, 0),
+            "{name}"
+        );
+    }
+}
+
+#[tokio::test]
+async fn warmup_confirms_idle_full_windows_even_when_the_saved_snapshot_is_old() {
+    let server = MockServer::start().await;
+    let store = Arc::new(MemoryAccountStore::default());
+    create_account(&store, "acct_idle").await;
+    seed_quota(
+        &store,
+        "acct_idle",
+        usage(short_window(
+            0,
+            Some(18_000),
+            Utc::now().timestamp() + 14_400,
+        )),
+        SystemTime::now() - Duration::from_secs(3_600),
+    )
+    .await;
+    Mock::given(method("GET"))
+        .and(path("/api/codex/usage"))
+        .respond_with(|_: &wiremock::Request| {
+            let mut quota = usage(short_window(
+                0,
+                Some(18_000),
+                Utc::now().timestamp() + 18_000,
+            ));
+            quota["plan_type"] = json!("team");
+            ResponseTemplate::new(200).set_body_json(quota)
+        })
+        .expect(2)
+        .mount(&server)
+        .await;
+    mount_completion(&server, 1).await;
+    let service = quota_service_with_base_url(&store, reqwest::Client::new(), server.uri());
+
+    let summary = service.execute_warmup("test-model").await.unwrap();
+
+    assert_eq!(
+        (summary.warmed_up, summary.skipped_active, summary.failed),
+        (1, 0, 0)
+    );
+}
+
+#[tokio::test]
+async fn warmup_refreshes_zero_percent_passive_windows_before_skipping() {
+    let server = MockServer::start().await;
+    let store = Arc::new(MemoryAccountStore::default());
+    create_account(&store, "acct_passive_zero").await;
+    let reset = Utc::now().timestamp() + 14_400;
+    seed_quota(
+        &store,
+        "acct_passive_zero",
+        usage(short_window(0, None, reset)),
+        SystemTime::now(),
+    )
+    .await;
+    Mock::given(method("GET"))
+        .and(path("/api/codex/usage"))
+        .respond_with(ResponseTemplate::new(200).set_body_json(usage(short_window(
+            0,
+            Some(14_400),
+            reset,
+        ))))
+        .expect(1)
+        .mount(&server)
+        .await;
+    mount_completion(&server, 0).await;
+    let service = quota_service_with_base_url(&store, reqwest::Client::new(), server.uri());
+
+    let summary = service.execute_warmup("test-model").await.unwrap();
+
+    assert_eq!(
+        (summary.warmed_up, summary.skipped_active, summary.failed),
+        (0, 1, 0)
+    );
+}
+
+#[tokio::test]
+async fn warmup_rechecks_a_just_activated_full_zero_percent_window() {
+    let server = MockServer::start().await;
+    let store = Arc::new(MemoryAccountStore::default());
+    create_account(&store, "acct_just_active").await;
+    let reset = Utc::now().timestamp() + 18_000;
+    let full = usage(short_window(0, Some(18_000), reset));
+    seed_quota(&store, "acct_just_active", full.clone(), SystemTime::now()).await;
+    Mock::given(method("GET"))
+        .and(path("/api/codex/usage"))
+        .respond_with(ResponseTemplate::new(200).set_body_json(full))
+        .with_priority(1)
+        .up_to_n_times(1)
+        .expect(1)
+        .mount(&server)
+        .await;
+    Mock::given(method("GET"))
+        .and(path("/api/codex/usage"))
+        .respond_with(ResponseTemplate::new(200).set_body_json(usage(short_window(
+            0,
+            Some(17_998),
+            reset,
+        ))))
+        .with_priority(2)
+        .expect(1)
+        .mount(&server)
+        .await;
+    mount_completion(&server, 0).await;
+    let service = quota_service_with_base_url(&store, reqwest::Client::new(), server.uri());
+
+    let summary = service.execute_warmup("test-model").await.unwrap();
+
+    assert_eq!(
+        (summary.warmed_up, summary.skipped_active, summary.failed),
+        (0, 1, 0)
+    );
+}
+
+#[tokio::test]
+async fn warmup_quota_verification_failure_does_not_abort_other_accounts() {
+    let server = MockServer::start().await;
+    let store = Arc::new(MemoryAccountStore::default());
+    create_account(&store, "acct_a_uncertain").await;
+    create_account(&store, "acct_b_ready").await;
+    seed_quota(
+        &store,
+        "acct_a_uncertain",
+        usage(short_window(0, None, Utc::now().timestamp() + 14_400)),
+        SystemTime::now(),
+    )
+    .await;
+    Mock::given(method("GET"))
+        .and(path("/api/codex/usage"))
+        .respond_with(ResponseTemplate::new(400))
+        .expect(1)
+        .mount(&server)
+        .await;
+    mount_completion(&server, 1).await;
+    let service = quota_service_with_base_url(&store, reqwest::Client::new(), server.uri());
+
+    let summary = service.execute_warmup("test-model").await.unwrap();
+
+    assert_eq!(
+        (summary.warmed_up, summary.skipped_active, summary.failed),
+        (1, 0, 1)
+    );
+}
+
+#[tokio::test]
+async fn warmup_respects_exhaustion_discovered_by_quota_verification() {
+    let server = MockServer::start().await;
+    let store = Arc::new(MemoryAccountStore::default());
+    create_account(&store, "acct_exhausted").await;
+    let reset = Utc::now().timestamp() + 14_400;
+    seed_quota(
+        &store,
+        "acct_exhausted",
+        usage(short_window(0, None, reset)),
+        SystemTime::now(),
+    )
+    .await;
+    let mut exhausted = usage(short_window(100, Some(14_400), reset));
+    exhausted["rate_limit"]["allowed"] = json!(false);
+    Mock::given(method("GET"))
+        .and(path("/api/codex/usage"))
+        .respond_with(ResponseTemplate::new(200).set_body_json(exhausted))
+        .expect(1)
+        .mount(&server)
+        .await;
+    mount_completion(&server, 0).await;
+    let service = quota_service_with_base_url(&store, reqwest::Client::new(), server.uri());
+
+    let summary = service.execute_warmup("test-model").await.unwrap();
+
+    assert_eq!(
+        (summary.warmed_up, summary.skipped_exhausted, summary.failed),
+        (0, 1, 0)
+    );
+}
+
+#[tokio::test]
+async fn warmup_bounds_quota_verification_when_activity_remains_unknown() {
+    let server = MockServer::start().await;
+    let store = Arc::new(MemoryAccountStore::default());
+    create_account(&store, "acct_unknown").await;
+    let quota = usage(short_window(0, None, Utc::now().timestamp() + 14_400));
+    seed_quota(&store, "acct_unknown", quota.clone(), SystemTime::now()).await;
+    Mock::given(method("GET"))
+        .and(path("/api/codex/usage"))
+        .respond_with(ResponseTemplate::new(200).set_body_json(quota))
+        .expect(2)
+        .mount(&server)
+        .await;
+    mount_completion(&server, 0).await;
+    let service = quota_service_with_base_url(&store, reqwest::Client::new(), server.uri());
+
+    let summary = service.execute_warmup("test-model").await.unwrap();
+
+    assert_eq!(
+        (summary.warmed_up, summary.skipped_active, summary.failed),
+        (0, 0, 1)
+    );
+}
 
 #[tokio::test]
 async fn warmup_consumes_completed_sse_and_persists_rate_limit_event() {

docs/api.md

@@ -2034,7 +2034,10 @@ HTTP 返回 `429`，`error.code` 为 `key_daily_budget_exceeded` 或 `key_weekly
 
 定时账号预热默认关闭。`accountWarmupScheduleTime` 使用部署时区中的 `HH:MM`，
 多个时段以逗号分隔，默认 `08:00`；`accountWarmupModel` 默认 `null`，开启前必须显式选择模型。
-任务面向可用的 Codex OAuth 账号，跳过周额度耗尽及五小时窗口距离重置仍超过 30 分钟的账号。
+任务面向可用的 Codex OAuth 账号，按账号级额度桶判断，跳过额度耗尽及已确认活跃、距离重置仍超过 30 分钟的五小时窗口。
+0% 不代表窗口空闲；已用额度或开始递减的上游倒计时用于确认活跃。距重置超过 30 分钟但缺少活跃证据时，
+最多回查两次，两次实时查询间隔 2 秒；0% 完整窗口连续随查询后移才视为空闲。
+回查失败或仍无法确认时计为失败，不发送预热。
 不存在的本地时刻跳过，重复时刻只执行较早一次；执行进度跨重启保留，时钟回拨不补跑已领取时刻之前的时段。
 只有收到响应成功终态才记为预热成功；预热不计入客户端业务用量
 

release/version.yaml

@@ -1,7 +1,7 @@
 version: 3.44.0
 published:
   '3':
-    stable: 3.43.1
-    rc: 3.43.1
-    beta: 3.43.1
-    alpha: 3.43.1
+    stable: 3.44.0
+    rc: 3.44.0
+    beta: 3.44.0
+    alpha: 3.44.0