From 47d08786ad5406af315bd8769f8d4a3f5e702dbc Mon Sep 17 00:00:00 2001 From: Zyy Date: Mon, 7 Sep 2026 04:47:30 +0800 Subject: [PATCH] fix(scheduling): retain valid quota signals across idle periods Keep OpenAI quota windows eligible for scheduling until their reset time, independently of the projection cache TTL. Expire undated observations and reset windows separately, and clear old signals when a fresh observation contains no active window. Record bounded candidate diagnostics with quota, concurrency, smart scores and affinity results. Cover idle 74%/96% accounts with one concurrent title request, independent window expiry, cache replacement and trace retention. Validation: workspace tests (1852 passed), rustfmt, and strict workspace Clippy with all targets and features. PostgreSQL/Redis integration services were not configured. Co-authored-by: Codex --- .../crates/gateway-core/src/account/mod.rs | 1 + .../gateway-core/src/account/selection.rs | 6 +- .../gateway-core/src/diagnostics/mod.rs | 1 + .../gateway-core/src/diagnostics/selection.rs | 73 +++++ .../gateway-core/src/diagnostics/trace.rs | 7 + .../crates/gateway-core/tests/account/mod.rs | 1 + .../tests/account/selection_diagnostics.rs | 92 ++++++ .../openai/src/credential/quota/snapshot.rs | 52 ++-- .../openai/src/credential/selector.rs | 7 +- .../openai/tests/credential/quota/mod.rs | 1 + .../tests/credential/quota/scheduling.rs | 277 ++++++++++++++++++ 11 files changed, 494 insertions(+), 24 deletions(-) create mode 100644 backend/crates/gateway-core/src/diagnostics/selection.rs create mode 100644 backend/crates/gateway-core/tests/account/selection_diagnostics.rs create mode 100644 backend/crates/providers/openai/tests/credential/quota/scheduling.rs diff --git a/backend/crates/gateway-core/src/account/mod.rs b/backend/crates/gateway-core/src/account/mod.rs index d8d9652a3..348306157 100644 --- a/backend/crates/gateway-core/src/account/mod.rs +++ b/backend/crates/gateway-core/src/account/mod.rs @@ -9,4 +9,5 @@ mod store; pub use error::CredentialError; pub use model::*; pub use selection::*; +pub(crate) use selection::{SMART_SCORE_TOLERANCE, smart_score}; pub use store::ProviderAccountStore; diff --git a/backend/crates/gateway-core/src/account/selection.rs b/backend/crates/gateway-core/src/account/selection.rs index 578de09b5..a3a9708b6 100644 --- a/backend/crates/gateway-core/src/account/selection.rs +++ b/backend/crates/gateway-core/src/account/selection.rs @@ -590,7 +590,7 @@ impl AccountSelector { }) } - fn scheduling_blocker( + pub(crate) fn scheduling_blocker( &self, candidate: &AccountCandidate, context: &AccountSelectionContext, @@ -643,7 +643,7 @@ const SMART_QUOTA_WEIGHT: f64 = 0.8; const SMART_FAILURE_WEIGHT: f64 = 1.0; const SMART_LATENCY_WEIGHT: f64 = 0.5; // 容忍 5 个百分点的单项负载/失败率差异，避免微小信号波动独占新会话。 -const SMART_SCORE_TOLERANCE: f64 = 0.05; +pub(crate) const SMART_SCORE_TOLERANCE: f64 = 0.05; // 首输出 10 秒时延迟得分减半；固定尺度不随其他候选账号变化。 const SMART_LATENCY_HALF_SCORE_MS: f64 = 10_000.0; @@ -677,7 +677,7 @@ fn select_smart_candidate<'a>( Some(ranked[index].0) } -fn smart_score(candidate: &AccountCandidate, default_concurrency: NonZeroU32) -> f64 { +pub(crate) fn smart_score(candidate: &AccountCandidate, default_concurrency: NonZeroU32) -> f64 { let load = 1.0 - capacity_utilization(candidate, default_concurrency).clamp(0.0, 1.0); let quota = candidate .signals diff --git a/backend/crates/gateway-core/src/diagnostics/mod.rs b/backend/crates/gateway-core/src/diagnostics/mod.rs index b5de9a0ef..ad3f346f8 100644 --- a/backend/crates/gateway-core/src/diagnostics/mod.rs +++ b/backend/crates/gateway-core/src/diagnostics/mod.rs @@ -1,6 +1,7 @@ //! 有界、请求局部的诊断证据；不参与路由、重试或客户端协议决策。 mod capture; +mod selection; mod stream; mod trace; diff --git a/backend/crates/gateway-core/src/diagnostics/selection.rs b/backend/crates/gateway-core/src/diagnostics/selection.rs new file mode 100644 index 000000000..7d146771e --- /dev/null +++ b/backend/crates/gateway-core/src/diagnostics/selection.rs @@ -0,0 +1,73 @@ +//! 复用选号时的输入，保留有界的候选评分与阻断原因。 + +use serde_json::json; + +use super::TraceContext; +use crate::account::{ + AccountCandidate, AccountSelection, AccountSelectionContext, AccountSelector, RotationStrategy, + SMART_SCORE_TOLERANCE, smart_score, +}; + +// 单个 trace 事件最多 4 KiB；优先保留实际选中的账号。 +const MAX_CANDIDATES: usize = 6; + +impl TraceContext { + /// 记录实际选号输入与结果；只复用评分规则，不参与调度决策。 + pub fn account_selection( + &self, + candidates: &[AccountCandidate], + context: &AccountSelectionContext, + selection: Option<&AccountSelection<'_>>, + ) { + if !self.is_enabled() { + return; + } + let selected = selection.as_ref().map(|s| s.candidate()); + let selected_id = selected.map(|c| c.account.id()); + let smart = context.policy.strategy() == RotationStrategy::Smart; + let observations = selected + .into_iter() + .chain( + candidates + .iter() + .filter(|c| Some(c.account.id()) != selected_id), + ) + .take(MAX_CANDIDATES) + .map(|candidate| { + let signals = &candidate.signals; + json!({ + "accountId": candidate.account.id().as_str(), + "weight": candidate.account.weight().get(), + "blocker": AccountSelector.scheduling_blocker(candidate, context) + .map(|blocker| format!("{blocker:?}")), + "inFlight": signals.in_flight, + "concurrencyLimit": candidate.account + .effective_concurrency(context.policy.max_concurrent_per_account()).get(), + "quotaRemainingBasisPoints": signals.quota_remaining_rank, + "quotaResetAtUnixMs": signals.quota_reset_at.and_then(|reset| + reset.duration_since(std::time::UNIX_EPOCH).ok() + ).map(|duration| duration.as_millis()), + "failureRateBasisPoints": signals.failure_rate_basis_points, + "firstOutputLatencyMs": signals.first_output_latency_ms, + "smartScore": smart.then(|| smart_score( + candidate, context.policy.max_concurrent_per_account() + )), + }) + }) + .collect::>(); + self.record( + "account.selection", + json!({ + "strategy": context.policy.strategy().as_str(), + "selectedAccountId": selected_id.map(|id| id.as_str()), + "preferredAccountId": context.preferred_account.as_ref().map(|id| id.as_str()), + "preferredResult": selection.as_ref().map(|s| format!("{:?}", s.preferred())), + "roundRobinCursor": context.round_robin_cursor, + "smartScoreTolerance": smart.then_some(SMART_SCORE_TOLERANCE), + "candidateCount": candidates.len(), + "omittedCandidates": candidates.len().saturating_sub(observations.len()), + "candidates": observations, + }), + ); + } +} diff --git a/backend/crates/gateway-core/src/diagnostics/trace.rs b/backend/crates/gateway-core/src/diagnostics/trace.rs index 01a056b96..624dbb122 100644 --- a/backend/crates/gateway-core/src/diagnostics/trace.rs +++ b/backend/crates/gateway-core/src/diagnostics/trace.rs @@ -65,6 +65,12 @@ struct TraceEvent { } impl TraceContext { + /// 是否已启用请求诊断，供调用方避免构造不会保存的事实。 + #[must_use] + pub const fn is_enabled(&self) -> bool { + self.state.is_some() + } + /// 一个模型请求只创建一次，所有 attempt 共享同一有界时间线。 #[must_use] pub fn new(request_id: &str) -> Self { @@ -306,6 +312,7 @@ impl TraceContext { !matches!( state.events[*index].stage, "attempt.started" + | "account.selection" | "account.selected" | "attempt.failed" | "retry.decided" diff --git a/backend/crates/gateway-core/tests/account/mod.rs b/backend/crates/gateway-core/tests/account/mod.rs index 7c15bbe43..c18cfd4e0 100644 --- a/backend/crates/gateway-core/tests/account/mod.rs +++ b/backend/crates/gateway-core/tests/account/mod.rs @@ -1,4 +1,5 @@ mod selection; +mod selection_diagnostics; use std::collections::{BTreeMap, BTreeSet}; use std::num::NonZeroU32; diff --git a/backend/crates/gateway-core/tests/account/selection_diagnostics.rs b/backend/crates/gateway-core/tests/account/selection_diagnostics.rs new file mode 100644 index 000000000..b28016e18 --- /dev/null +++ b/backend/crates/gateway-core/tests/account/selection_diagnostics.rs @@ -0,0 +1,92 @@ +use gateway_core::account::{AccountSelector, RotationStrategy}; +use gateway_core::diagnostics::TraceContext; +use serde_json::json; + +use super::{candidate, candidate_with_concurrency, context}; + +#[test] +fn selection_trace_should_distinguish_load_quota_and_affinity() { + let mut candidates = [ + candidate_with_concurrency("acct_74", 1, 10), + candidate_with_concurrency("acct_96", 0, 10), + ]; + candidates[0].signals.quota_remaining_rank = Some(2600); + candidates[1].signals.quota_remaining_rank = Some(400); + let mut context = context(RotationStrategy::Smart); + let trace = TraceContext::new("req_smart_trace"); + let selection = AccountSelector.select(&candidates, &context); + trace.account_selection(&candidates, &context, selection.as_ref()); + context.preferred_account = Some(candidates[1].account.id().clone()); + let selection = AccountSelector.select(&candidates, &context); + trace.account_selection(&candidates, &context, selection.as_ref()); + + let snapshot = trace.snapshot().expect("trace"); + let first = &snapshot["events"][0]["data"]; + let second = &snapshot["events"][1]["data"]; + assert_eq!(first["selectedAccountId"], "acct_74"); + assert_eq!(first["preferredResult"], "NotRequested"); + let busy = &first["candidates"][0]; + assert_eq!(busy["inFlight"], 1); + assert_eq!(busy["concurrencyLimit"], 10); + assert_eq!(busy["quotaRemainingBasisPoints"], 2600); + assert!(busy["blocker"].is_null()); + assert!( + busy["smartScore"].as_f64().expect("score") + > first["candidates"][1]["smartScore"] + .as_f64() + .expect("score") + ); + assert_eq!(second["selectedAccountId"], "acct_96"); + assert_eq!(second["preferredResult"], "Hit"); +} + +#[test] +fn selection_trace_should_bound_candidates_and_preserve_the_selected_account() { + let mut candidates = (0..30) + .map(|index| candidate(&format!("acct_{index}"), 0, None)) + .collect::>(); + candidates.push(candidate("acct_best", 0, Some(10_000))); + let trace = TraceContext::new("req_many_candidates"); + // 选号位于首部保留区之外，也应在长流的常规事件淘汰后保留。 + for _ in 0..10 { + trace.record("test.event", json!({})); + } + let context = context(RotationStrategy::Smart); + let selection = AccountSelector.select(&candidates, &context); + trace.account_selection(&candidates, &context, selection.as_ref()); + for index in 0..400 { + trace.record("test.event", json!({"index": index})); + } + + let snapshot = trace.snapshot().expect("trace"); + let event = snapshot["events"] + .as_array() + .expect("events") + .iter() + .find(|event| event["stage"] == "account.selection") + .expect("selection survives event eviction"); + let data = &event["data"]; + let recorded = data["candidates"].as_array().expect("bounded candidates"); + assert!(recorded.len() < candidates.len()); + assert_eq!(recorded[0]["accountId"], "acct_best"); + assert!(recorded[1]["quotaRemainingBasisPoints"].is_null()); + assert_eq!( + recorded.len() as u64 + data["omittedCandidates"].as_u64().expect("omitted"), + candidates.len() as u64 + ); + assert!(data.to_string().len() < 4096); +} + +#[test] +fn selection_trace_should_explain_when_all_accounts_are_busy() { + let candidates = [candidate_with_concurrency("acct_busy", 10, 10)]; + let trace = TraceContext::new("req_no_candidate"); + let context = context(RotationStrategy::Smart); + let selection = AccountSelector.select(&candidates, &context); + assert!(selection.is_none()); + trace.account_selection(&candidates, &context, selection.as_ref()); + let snapshot = trace.snapshot().expect("trace"); + let data = &snapshot["events"][0]["data"]; + assert!(data["selectedAccountId"].is_null()); + assert_eq!(data["candidates"][0]["blocker"], "ConcurrencyLimit"); +} diff --git a/backend/crates/providers/openai/src/credential/quota/snapshot.rs b/backend/crates/providers/openai/src/credential/quota/snapshot.rs index dfccd3353..caa6a34bd 100644 --- a/backend/crates/providers/openai/src/credential/quota/snapshot.rs +++ b/backend/crates/providers/openai/src/credential/quota/snapshot.rs @@ -236,25 +236,36 @@ pub(crate) fn quota_snapshot_from_observation( Some(snapshot) } -/// 调度投影缓存 TTL：`min(常规 TTL, 最近 active reset - now)`。 -/// reset 到期时缓存自然失效，不固定多等一个常规 TTL 周期。 +/// 缓存定期重读持久化事实，额度窗口独立按 reset 到期。 +/// 正常账号依赖被动同步，空闲超过缓存 TTL 不能把尚未重置的额度抹成未知。 pub(crate) fn quota_projection_ttl(snapshot: &CodexAccountQuotaSnapshot) -> Option { - let age = SystemTime::now() - .duration_since(snapshot.observed_at()) - .unwrap_or(Duration::ZERO); - let normal_ttl = QUOTA_SCHEDULING_TTL.checked_sub(age)?; let now = SystemTime::now(); - let reset_ttl = snapshot + snapshot .windows() .iter() - .filter_map(|window| window.reset_at()) - .map(SystemTime::from) - .filter(|reset| *reset > now) - .map(|reset| reset.duration_since(now).unwrap_or(Duration::ZERO)) - .min(); - let ttl = match reset_ttl { - Some(reset_ttl) if reset_ttl < normal_ttl => reset_ttl, - _ => normal_ttl, + .filter_map(|window| quota_window_ttl(window, snapshot.observed_at(), now)) + .min() + // 新观测即使没有有效窗口，也要用“未知”替换此前缓存的排序信号。 + .or_else(|| { + QUOTA_SCHEDULING_TTL.checked_sub( + now.duration_since(snapshot.observed_at()) + .unwrap_or(Duration::ZERO), + ) + }) + .filter(|ttl| !ttl.is_zero()) + .map(|ttl| ttl.min(QUOTA_SCHEDULING_TTL)) +} + +fn quota_window_ttl( + window: &CodexQuotaWindow, + observed_at: SystemTime, + now: SystemTime, +) -> Option { + let ttl = match window.reset_at() { + Some(reset_at) => SystemTime::from(reset_at).duration_since(now).ok()?, + // 没有明确窗口边界时，不能无限延长旧观测的有效期。 + None => QUOTA_SCHEDULING_TTL + .checked_sub(now.duration_since(observed_at).unwrap_or(Duration::ZERO))?, }; (!ttl.is_zero()).then_some(ttl) } @@ -273,12 +284,13 @@ pub(crate) fn quota_scheduling_signals( let mut remaining_rank: Option = None; let mut reset_at: Option = None; for window in &snapshot.windows { + if quota_window_ttl(window, snapshot.observed_at(), now).is_none() { + continue; + } let window_reset_at = window.reset_at().map(SystemTime::from); - let observation_expired = window_reset_at.is_some_and(|reset_at| reset_at <= now); - if !observation_expired - && let Some(used) = window - .used_percent() - .filter(|used| used.is_finite() && (0.0..=100.0).contains(used)) + if let Some(used) = window + .used_percent() + .filter(|used| used.is_finite() && (0.0..=100.0).contains(used)) { let rank = ((100.0 - used) * 100.0).round() as u64; remaining_rank = Some(remaining_rank.map_or(rank, |current| current.min(rank))); diff --git a/backend/crates/providers/openai/src/credential/selector.rs b/backend/crates/providers/openai/src/credential/selector.rs index c6040607c..0ba7228d1 100644 --- a/backend/crates/providers/openai/src/credential/selector.rs +++ b/backend/crates/providers/openai/src/credential/selector.rs @@ -512,7 +512,12 @@ impl CodexCredentialSelector { account_scope: request.attempt.account_scope().cloned(), }; let capacity = AccountSelector.capacity_snapshot(&candidates, &context); - let Some(selection) = AccountSelector.select(&candidates, &context) else { + let selection = AccountSelector.select(&candidates, &context); + request + .attempt + .trace() + .account_selection(&candidates, &context, selection.as_ref()); + let Some(selection) = selection else { return match shortest_retry { Some(retry_after) => Err(CredentialSelectionError::CapacityUnavailable { retry_after: Some(retry_after), diff --git a/backend/crates/providers/openai/tests/credential/quota/mod.rs b/backend/crates/providers/openai/tests/credential/quota/mod.rs index 48dcc25f2..f05f4070d 100644 --- a/backend/crates/providers/openai/tests/credential/quota/mod.rs +++ b/backend/crates/providers/openai/tests/credential/quota/mod.rs @@ -1,6 +1,7 @@ //! OpenAI 额度事实边界与展示快照回归。 mod recovery; +mod scheduling; mod slots; mod snapshot; diff --git a/backend/crates/providers/openai/tests/credential/quota/scheduling.rs b/backend/crates/providers/openai/tests/credential/quota/scheduling.rs new file mode 100644 index 000000000..98764671b --- /dev/null +++ b/backend/crates/providers/openai/tests/credential/quota/scheduling.rs @@ -0,0 +1,277 @@ +//! 持久化额度的有效期与短暂并发占用共同参与选号。 + +use std::collections::BTreeSet; +use std::num::NonZeroU32; +use std::sync::Arc; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use gateway_core::account::{ + AccountCandidate, AccountConcurrencyLimit, AccountEligibilityPolicy, AccountRuntimeSignals, + AccountSelectionContext, AccountSelectionPolicy, AccountSelector, AccountWeight, + OpaqueProviderData, ProviderAccount, ProviderAccountStore as _, QuotaAccessState, + QuotaEvidence, QuotaObservation, QuotaState, RotationStrategy, +}; +use serde_json::{Value, json}; + +use super::{MemoryAccountStore, create_account, quota_service}; + +async fn persist_snapshot( + store: &Arc, + id: &str, + observed_at: SystemTime, + rate_limit: Value, +) -> ProviderAccount { + create_account(store, id).await; + let account = store.account(id).expect("created account"); + store + .compare_and_swap_quota(QuotaObservation { + account_id: account.id().clone(), + expected_revision: account.revision(), + quota: OpaqueProviderData::new( + json!({"rate_limit": rate_limit}) + .as_object() + .expect("quota object") + .clone(), + ), + observed_at, + state: QuotaState::allowed(observed_at), + }) + .await + .expect("persist quota"); + store.account(id).expect("persisted account") +} + +fn timestamp(time: SystemTime) -> u64 { + time.duration_since(UNIX_EPOCH).expect("epoch").as_secs() +} + +#[tokio::test] +async fn old_weekly_quota_should_outweigh_one_in_flight_title_request() { + let store = Arc::new(MemoryAccountStore::default()); + let now = SystemTime::now(); + let reset_at = timestamp(now + Duration::from_secs(6 * 3600)); + let mut accounts = Vec::new(); + for (id, used_percent, age_minutes) in [("acct_74", 74, 41), ("acct_96", 96, 16)] { + accounts.push( + persist_snapshot( + &store, + id, + now - Duration::from_secs(age_minutes * 60), + json!({ + "allowed": true, + "primary_window": { + "used_percent": used_percent, + "reset_at": reset_at, + "limit_window_seconds": 604_800 + } + }), + ) + .await + .with_scheduling( + Some(AccountConcurrencyLimit::new(10).expect("concurrency")), + AccountWeight::DEFAULT, + ), + ); + } + let service = quota_service(&store); + service.prepare_scheduling(&accounts).await; + let candidates = accounts + .into_iter() + .map(|account| AccountCandidate { + signals: AccountRuntimeSignals { + in_flight: u32::from(account.id().as_str() == "acct_74"), + last_started_at: None, + quota_reset_at: None, + quota_remaining_rank: None, + rate_limited_until: None, + failure_rate_basis_points: None, + first_output_latency_ms: None, + } + .with_provider_quota(service.scheduling_signals(&account)), + account, + }) + .collect::>(); + for cursor in 0..20 { + let context = AccountSelectionContext { + policy: AccountSelectionPolicy::new( + RotationStrategy::Smart, + NonZeroU32::new(3).expect("default concurrency"), + Duration::ZERO, + ), + now, + excluded_accounts: BTreeSet::new(), + preferred_account: None, + preferred_account_overrides_weight: true, + round_robin_cursor: cursor, + eligibility: AccountEligibilityPolicy::Enforce, + account_scope: None, + }; + let selected = AccountSelector + .select(&candidates, &context) + .expect("available account"); + assert_eq!(selected.candidate().account.id().as_str(), "acct_74"); + } +} + +#[tokio::test] +async fn old_quota_should_drop_reset_short_window_and_keep_weekly_window() { + let store = Arc::new(MemoryAccountStore::default()); + let now = SystemTime::now(); + let weekly_reset = now + Duration::from_secs(6 * 3600); + let account = persist_snapshot( + &store, + "acct_mixed_resets", + now - Duration::from_secs(41 * 60), + json!({ + "primary_window": { + "used_percent": 99, + "reset_at": timestamp(now - Duration::from_secs(60)), + "limit_window_seconds": 18_000 + }, + "secondary_window": { + "used_percent": 74, + "reset_at": timestamp(weekly_reset), + "limit_window_seconds": 604_800 + } + }), + ) + .await; + let service = quota_service(&store); + service + .read_account(account.id()) + .await + .expect("read quota"); + let signals = service.scheduling_signals(&account).expect("weekly quota"); + assert_eq!( + (signals.remaining_rank(), signals.reset_at().map(timestamp)), + (Some(2600), Some(timestamp(weekly_reset))) + ); +} + +#[tokio::test] +async fn undated_quota_should_expire_independently_of_dated_weekly_quota() { + for (age_minutes, expected_remaining) in [(1, 100), (41, 2600)] { + let store = Arc::new(MemoryAccountStore::default()); + let now = SystemTime::now(); + let account = persist_snapshot( + &store, + "acct_undated_window", + now - Duration::from_secs(age_minutes * 60), + json!({ + "primary_window": {"used_percent": 99}, + "secondary_window": { + "used_percent": 74, + "reset_at": timestamp(now + Duration::from_secs(6 * 3600)), + "limit_window_seconds": 604_800 + } + }), + ) + .await; + let service = quota_service(&store); + service + .prepare_scheduling(std::slice::from_ref(&account)) + .await; + assert_eq!( + service + .scheduling_signals(&account) + .and_then(|s| s.remaining_rank()), + Some(expected_remaining), + "observation age: {age_minutes} minutes" + ); + } +} + +#[tokio::test] +async fn old_undated_quota_should_remain_unknown_after_cache_hydration() { + let store = Arc::new(MemoryAccountStore::default()); + let account = persist_snapshot( + &store, + "acct_undated_only", + SystemTime::now() - Duration::from_secs(41 * 60), + json!({"primary_window": {"used_percent": 74}}), + ) + .await; + let service = quota_service(&store); + for _ in 0..2 { + service + .prepare_scheduling(std::slice::from_ref(&account)) + .await; + assert_eq!(service.scheduling_signals(&account), None); + } + assert_eq!(store.quota_reads(), 1); +} + +#[tokio::test] +async fn fresh_reset_observation_should_clear_previous_scheduling_signals() { + let store = Arc::new(MemoryAccountStore::default()); + let now = SystemTime::now(); + let account = persist_snapshot( + &store, + "acct_fresh_reset", + now, + json!({"primary_window": { + "used_percent": 74, + "reset_at": timestamp(now + Duration::from_secs(3600)) + }}), + ) + .await; + let service = quota_service(&store); + service + .prepare_scheduling(std::slice::from_ref(&account)) + .await; + assert_eq!( + service + .scheduling_signals(&account) + .and_then(|s| s.remaining_rank()), + Some(2600) + ); + + let headers = vec![ + ("x-codex-active-limit".to_owned(), "codex".to_owned()), + ("x-codex-primary-used-percent".to_owned(), "74".to_owned()), + ( + "x-codex-primary-reset-at".to_owned(), + timestamp(now - Duration::from_secs(60)).to_string(), + ), + ]; + assert!( + service + .synchronize_passive_headers(&account, &headers) + .await + .expect("reset observation") + ); + assert_eq!(service.scheduling_signals(&account), None); +} + +#[tokio::test] +async fn reset_window_should_not_clear_confirmed_exhaustion() { + let store = Arc::new(MemoryAccountStore::default()); + let now = SystemTime::now(); + let reset_at = now - Duration::from_secs(60); + let account = persist_snapshot( + &store, + "acct_exhausted_reset", + now - Duration::from_secs(41 * 60), + json!({"primary_window": {"used_percent": 100, "reset_at": timestamp(reset_at)}}), + ) + .await; + super::persist_quota_state( + &store, + &account, + QuotaState::exhausted(QuotaEvidence::ProviderDenied, now, Some(reset_at)), + ) + .await; + let service = quota_service(&store); + service + .prepare_scheduling(std::slice::from_ref(&account)) + .await; + assert_eq!(service.scheduling_signals(&account), None); + assert_eq!( + store + .account("acct_exhausted_reset") + .expect("account") + .quota() + .access(), + QuotaAccessState::Exhausted + ); +}