fix(openai): recover quota after same-window usage drops (#95)

OpenAI 周窗口的 used_percent 可以在同一窗口内回落（滚动计量），而 reset_at 固定到窗口边界才变化。恢复规则要求「reset 前进 + 用量低于 10%」，两个条件在滑动周窗口上都无法成立：耗尽账号被钉在旧 reset 日期（最长一周）无法参与调度，尽管 /usage 已显式报告 allowed=true / limit_reached=false 且 5h 窗口全新。新增第二条解除路径：同一窗口（reset 未滚动）内连续两次观测都未触顶即解除。候选记录持久化在 _quota_recovery，跨 worker 周期与重启累积；连续两次的要求保持对「耗尽后立刻拉到的旧快照」竞态的防护（单次旧快照撑不过两次观测）。原有「reset 前进 + 用量<10%」快路径不变。
2026-11-06 08:06:19 +00:00 · 2026-09-15 15:38:46 +08:00
parent 817b0f2d6d
commit dfabf0f9dd
2 changed files with 142 additions and 12 deletions
@@ -18,6 +18,10 @@ pub(super) struct QuotaRecovery {
    // 绑定本次耗尽事实，避免真实推理成功后再次耗尽时复用上次的恢复进度。
    exhausted_at_micros: i64,
    pending: BTreeMap<String, Option<DateTime<Utc>>>,
+    // 已连续观测到"未触顶"的待解除窗口；第二次确认后解除。旧文档没有该
+    // 字段时反序列化为空集，从当前耗尽事实重新积累证据。
+    #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
+    candidates: BTreeMap<String, ()>,
 }

 impl QuotaRecovery {
@@ -59,23 +63,44 @@ impl QuotaRecovery {
        Self {
            exhausted_at_micros: exhaustion_time(state),
            pending,
+            candidates: BTreeMap::new(),
        }
    }

    fn observe(&mut self, refreshed: &CodexAccountQuotaSnapshot) -> bool {
        let had_baseline = !self.pending.is_empty();
+        let mut candidates = BTreeMap::new();
        for window in refreshed.windows().iter().filter(|w| w.is_account_wide()) {
-            if let Some(previous_reset) = self.pending.get_mut(window.key()) {
-                if window_reset_recovered(window, *previous_reset) {
-                    self.pending.remove(window.key());
-                } else if previous_reset.is_none() {
-                    *previous_reset = window.reset_at();
+            let key = window.key();
+            if let Some(previous_reset) = self.pending.get(key).copied() {
+                if window_reset_recovered(window, previous_reset) {
+                    self.pending.remove(key);
+                    continue;
+                }
+                if previous_reset.is_none() {
+                    // 首次观测到该窗口只锚定 reset 基线，不构成解除证据。
+                    self.pending.insert(key.to_owned(), window.reset_at());
+                    continue;
+                }
+                // 同一窗口内用量回落（reset 不滚动）时"reset 前进"永远不会
+                // 成立，滑动窗口的正常用量也长期高于低用量阈值；以上游连续
+                // 两次观测都报告未触顶为解除证据。单次观测无法排除耗尽后
+                // 立刻拉到的旧快照（reset 未滚动、用量尚未触顶）。
+                let rolled_forward = window
+                    .reset_at()
+                    .is_some_and(|current| Some(current) > previous_reset);
+                if !rolled_forward && !window.limit_reached() {
+                    if self.candidates.contains_key(key) {
+                        self.pending.remove(key);
+                    } else {
+                        candidates.insert(key.to_owned(), ());
+                    }
                }
            } else if window.limit_reached() {
-                self.pending
-                    .insert(window.key().to_owned(), window.reset_at());
+                self.pending.insert(key.to_owned(), window.reset_at());
            }
        }
+        self.candidates = candidates;
        had_baseline && self.pending.is_empty()
    }
 }
@@ -85,15 +110,24 @@ fn exhaustion_time(state: QuotaState) -> i64 {
 }

 fn window_reset_recovered(window: &CodexQuotaWindow, previous: Option<DateTime<Utc>>) -> bool {
-    previous
+    // reset 必须前进：防止耗尽后立刻拉到的旧快照（reset 未滚动）误解锁。
+    if !previous
        .zip(window.reset_at())
        .is_some_and(|(previous, current)| current > previous)
-        && window
-            .used_percent()
-            .is_some_and(|used| used < RESET_RECOVERY_MAX_USED_PERCENT)
+    {
+        return false;
+    }
+    // reset 前进说明已进入新窗口，新窗口用量应接近零；高用量说明 reset 基线
+    // 不可信（例如滑动窗口），保持锁定，等待"同窗口用量回落"的连续观测解除。
+    window
+        .used_percent()
+        .is_some_and(|used| used < RESET_RECOVERY_MAX_USED_PERCENT)
 }

-/// 统一手动刷新和 worker 的恢复规则；上游 allowed 不参与已耗尽窗口的解除判断。
+/// 统一手动刷新和 worker 的恢复规则。已耗尽窗口的解除有两条证据路径：
+/// reset 前进且用量低于低用量阈值（窗口已滚动），或同一窗口内连续两次观测
+/// 都未触顶（滑动窗口的用量回落，reset 不滚动）。后者防止账号被钉在旧
+/// reset 日期上；连续两次的要求排除耗尽后立刻拉到的旧快照。
 pub(super) fn reconcile_refresh(
    current: QuotaState,
    refreshed: &mut CodexAccountQuotaSnapshot,
@@ -561,6 +561,102 @@ async fn an_unknown_slot_establishes_a_baseline_before_it_can_recover() {
    }
 }

+#[tokio::test]
+async fn weekly_usage_falling_back_within_the_same_window_recovers_after_two_observations() {
+    // 滑动周窗口的用量可以在同一窗口内回落：reset 不滚动、用量从 100% 降到
+    // 正常水平。此时"reset 前进 + 低用量"永远不会成立，解除依赖连续两次
+    // 观测都未触顶；单次观测无法排除耗尽后立刻拉到的旧快照，保持锁定。
+    let store = Arc::new(MemoryAccountStore::default());
+    create_account(&store, "acct_weekly_fallout").await;
+    let account = store.account("acct_weekly_fallout").expect("account");
+    let server = MockServer::start().await;
+    let service = quota_service_with_base_url(&store, reqwest::Client::new(), server.uri());
+    let mut old = usage((100, SHORT_RESET), (100, WEEK_RESET));
+    old["rate_limit"]["allowed"] = json!(false);
+    old["rate_limit"]["limit_reached"] = json!(true);
+    mount_usage(&server, old).await;
+    service
+        .refresh_account(account.id())
+        .await
+        .expect("seed exhausted quota");
+
+    // 5h 窗口滚动前进，周窗口 reset 不变但用量回落到正常水平：只标记候选。
+    let mut first_observation = usage((0, SHORT_RESET + 18_000), (18, WEEK_RESET));
+    first_observation["rate_limit"]["allowed"] = json!(true);
+    mount_usage(&server, first_observation).await;
+    let first = service
+        .refresh_account(account.id())
+        .await
+        .expect("first fallout observation");
+    assert!(first.quota().is_exhausted());
+
+    // 第二次连续观测仍未触顶：解除周窗口。
+    let mut second_observation = usage((9, SHORT_RESET + 36_000), (18, WEEK_RESET));
+    second_observation["rate_limit"]["allowed"] = json!(true);
+    mount_usage(&server, second_observation).await;
+    let second = service
+        .refresh_account(account.id())
+        .await
+        .expect("second fallout observation");
+    assert_eq!(second.quota().access(), QuotaAccessState::Allowed);
+}
+
+#[tokio::test]
+async fn a_single_same_window_fallout_observation_cannot_break_the_reset_baseline() {
+    // 连续性中断（观测到重新触顶）后必须重新积累两次未触顶证据。
+    let store = Arc::new(MemoryAccountStore::default());
+    create_account(&store, "acct_weekly_fallout_gap").await;
+    let account = store.account("acct_weekly_fallout_gap").expect("account");
+    let server = MockServer::start().await;
+    let service = quota_service_with_base_url(&store, reqwest::Client::new(), server.uri());
+    let mut old = usage((100, SHORT_RESET), (100, WEEK_RESET));
+    old["rate_limit"]["allowed"] = json!(false);
+    mount_usage(&server, old).await;
+    service
+        .refresh_account(account.id())
+        .await
+        .expect("seed exhausted quota");
+
+    mount_usage(&server, usage((0, SHORT_RESET + 18_000), (18, WEEK_RESET))).await;
+    assert!(
+        service
+            .refresh_account(account.id())
+            .await
+            .expect("first candidate")
+            .quota()
+            .is_exhausted()
+    );
+    // 观测到重新触顶，候选证据清零。
+    mount_usage(&server, usage((0, SHORT_RESET + 36_000), (100, WEEK_RESET))).await;
+    assert!(
+        service
+            .refresh_account(account.id())
+            .await
+            .expect("reached again")
+            .quota()
+            .is_exhausted()
+    );
+    mount_usage(&server, usage((0, SHORT_RESET + 54_000), (18, WEEK_RESET))).await;
+    assert!(
+        service
+            .refresh_account(account.id())
+            .await
+            .expect("candidate restarts")
+            .quota()
+            .is_exhausted()
+    );
+    mount_usage(&server, usage((0, SHORT_RESET + 72_000), (18, WEEK_RESET))).await;
+    assert_eq!(
+        service
+            .refresh_account(account.id())
+            .await
+            .expect("second consecutive observation")
+            .quota()
+            .access(),
+        QuotaAccessState::Allowed
+    );
+}
+
 #[tokio::test]
 async fn monthly_and_arbitrary_periods_use_the_same_recovery_rule() {
    for seconds in [18_000, 86_400, 604_800, 864_000, 2_592_000, 7_776_000] {